openAgent

1import { LayerNode } from "@opencode-ai/core/effect/layer-node"
2import path from "path"
3import { Global } from "@opencode-ai/core/global"
4import { FSUtil } from "@opencode-ai/core/fs-util"
5import { Effect, Exit, Layer, Option, RcMap, Schema, Context, TxReentrantLock } from "effect"
6import { NonNegativeInt } from "@opencode-ai/core/schema"
7import { Git } from "@/git"
8 
9type Migration = (dir: string, fs: FSUtil.Interface, git: Git.Interface) => Effect.Effect<void, FSUtil.Error>
10 
11export class NotFoundError extends Schema.TaggedErrorClass<NotFoundError>()("NotFoundError", {
12  message: Schema.String,
13}) {
14  static isInstance(input: unknown): input is NotFoundError {
15    return input instanceof NotFoundError
16  }
17}
18 
19export type Error = FSUtil.Error | NotFoundError
20 
21const RootFile = Schema.Struct({
22  path: Schema.optional(
23    Schema.Struct({
24      root: Schema.optional(Schema.String),
25    }),
26  ),
27})
28 
29const SessionFile = Schema.Struct({
30  id: Schema.String,
31})
32 
33const MessageFile = Schema.Struct({
34  id: Schema.String,
35})
36 
37const DiffFile = Schema.Struct({
38  additions: NonNegativeInt,
39  deletions: NonNegativeInt,
40})
41 
42const SummaryFile = Schema.Struct({
43  id: Schema.String,
44  projectID: Schema.String,
45  summary: Schema.Struct({ diffs: Schema.Array(DiffFile) }),
46})
47 
48const decodeRoot = Schema.decodeUnknownOption(RootFile)
49const decodeSession = Schema.decodeUnknownOption(SessionFile)
50const decodeMessage = Schema.decodeUnknownOption(MessageFile)
51const decodeSummary = Schema.decodeUnknownOption(SummaryFile)
52 
53export interface Interface {
54  readonly remove: (key: string[]) => Effect.Effect<void, FSUtil.Error>
55  readonly read: <T>(key: string[]) => Effect.Effect<T, Error>
56  readonly update: <T>(key: string[], fn: (draft: T) => void) => Effect.Effect<T, Error>
57  readonly write: <T>(key: string[], content: T) => Effect.Effect<void, FSUtil.Error>
58  readonly list: (prefix: string[]) => Effect.Effect<string[][], FSUtil.Error>
59}
60 
61export class Service extends Context.Service<Service, Interface>()("@opencode/Storage") {}
62 
63function file(dir: string, key: string[]) {
64  return path.join(dir, ...key) + ".json"
65}
66 
67function missing(err: unknown) {
68  if (!err || typeof err !== "object") return false
69  if ("code" in err && err.code === "ENOENT") return true
70  if ("reason" in err && err.reason && typeof err.reason === "object" && "_tag" in err.reason) {
71    return err.reason._tag === "NotFound"
72  }
73  return false
74}
75 
76function parseMigration(text: string) {
77  const value = Number.parseInt(text, 10)
78  return Number.isNaN(value) ? 0 : value
79}
80 
81const MIGRATIONS: Migration[] = [
82  Effect.fn("Storage.migration.1")(function* (dir: string, fs: FSUtil.Interface, git: Git.Interface) {
83    const project = path.resolve(dir, "../project")
84    if (!(yield* fs.isDir(project))) return
85    const projectDirs = yield* fs.glob("*", {
86      cwd: project,
87      include: "all",
88    })
89    for (const projectDir of projectDirs) {
90      const full = path.join(project, projectDir)
91      if (!(yield* fs.isDir(full))) continue
92      yield* Effect.logInfo(`migrating project ${projectDir}`)
93      let projectID = projectDir
94      let worktree = "/"
95 
96      if (projectID !== "global") {
97        for (const msgFile of yield* fs.glob("storage/session/message/*/*.json", {
98          cwd: full,
99          absolute: true,
100        })) {
101          const json = decodeRoot(yield* fs.readJson(msgFile), { onExcessProperty: "preserve" })
102          const root = Option.isSome(json) ? json.value.path?.root : undefined
103          if (!root) continue
104          worktree = root
105          break
106        }
107        if (!worktree) continue
108        if (!(yield* fs.isDir(worktree))) continue
109        const result = yield* git.run(["rev-list", "--max-parents=0", "--all"], {
110          cwd: worktree,
111        })
112        const [id] = result
113          .text()
114          .split("\n")
115          .filter(Boolean)
116          .map((x) => x.trim())
117          .toSorted()
118        if (!id) continue
119        projectID = id
120 
121        yield* fs.writeWithDirs(
122          path.join(dir, "project", projectID + ".json"),
123          JSON.stringify(
124            {
125              id,
126              vcs: "git",
127              worktree,
128              time: {
129                created: Date.now(),
130                initialized: Date.now(),
131              },
132            },
133            null,
134            2,
135          ),
136        )
137 
138        yield* Effect.logInfo(`migrating sessions for project ${projectID}`)
139        for (const sessionFile of yield* fs.glob("storage/session/info/*.json", {
140          cwd: full,
141          absolute: true,
142        })) {
143          const dest = path.join(dir, "session", projectID, path.basename(sessionFile))
144          yield* Effect.logInfo("copying", { sessionFile, dest })
145          const session = yield* fs.readJson(sessionFile)
146          const info = decodeSession(session, { onExcessProperty: "preserve" })
147          yield* fs.writeWithDirs(dest, JSON.stringify(session, null, 2))
148          if (Option.isNone(info)) continue
149          yield* Effect.logInfo(`migrating messages for session ${info.value.id}`)
150          for (const msgFile of yield* fs.glob(`storage/session/message/${info.value.id}/*.json`, {
151            cwd: full,
152            absolute: true,
153          })) {
154            const next = path.join(dir, "message", info.value.id, path.basename(msgFile))
155            yield* Effect.logInfo("copying", {
156              msgFile,
157              dest: next,
158            })
159            const message = yield* fs.readJson(msgFile)
160            const item = decodeMessage(message, { onExcessProperty: "preserve" })
161            yield* fs.writeWithDirs(next, JSON.stringify(message, null, 2))
162            if (Option.isNone(item)) continue
163 
164            yield* Effect.logInfo(`migrating parts for message ${item.value.id}`)
165            for (const partFile of yield* fs.glob(`storage/session/part/${info.value.id}/${item.value.id}/*.json`, {
166              cwd: full,
167              absolute: true,
168            })) {
169              const out = path.join(dir, "part", item.value.id, path.basename(partFile))
170              const part = yield* fs.readJson(partFile)
171              yield* Effect.logInfo("copying", {
172                partFile,
173                dest: out,
174              })
175              yield* fs.writeWithDirs(out, JSON.stringify(part, null, 2))
176            }
177          }
178        }
179      }
180    }
181  }),
182  Effect.fn("Storage.migration.2")(function* (dir: string, fs: FSUtil.Interface) {
183    for (const item of yield* fs.glob("session/*/*.json", {
184      cwd: dir,
185      absolute: true,
186    })) {
187      const raw = yield* fs.readJson(item)
188      const session = decodeSummary(raw, { onExcessProperty: "preserve" })
189      if (Option.isNone(session)) continue
190      const diffs = session.value.summary.diffs
191      yield* fs.writeWithDirs(
192        path.join(dir, "session_diff", session.value.id + ".json"),
193        JSON.stringify(diffs, null, 2),
194      )
195      yield* fs.writeWithDirs(
196        path.join(dir, "session", session.value.projectID, session.value.id + ".json"),
197        JSON.stringify(
198          {
199            ...(raw as Record<string, unknown>),
200            summary: {
201              additions: diffs.reduce((sum, x) => sum + x.additions, 0),
202              deletions: diffs.reduce((sum, x) => sum + x.deletions, 0),
203            },
204          },
205          null,
206          2,
207        ),
208      )
209    }
210  }),
211]
212 
213const layer = Layer.effect(
214  Service,
215  Effect.gen(function* () {
216    const fs = yield* FSUtil.Service
217    const git = yield* Git.Service
218    const locks = yield* RcMap.make({
219      lookup: () => TxReentrantLock.make(),
220      idleTimeToLive: 0,
221    })
222    const state = yield* Effect.cached(
223      Effect.gen(function* () {
224        const dir = path.join(Global.Path.data, "storage")
225        const marker = path.join(dir, "migration")
226        const migration = yield* fs.readFileString(marker).pipe(
227          Effect.map(parseMigration),
228          Effect.catchIf(missing, () => Effect.succeed(0)),
229          Effect.orElseSucceed(() => 0),
230        )
231        for (let i = migration; i < MIGRATIONS.length; i++) {
232          yield* Effect.logInfo("running migration", { index: i })
233          const step = MIGRATIONS[i]!
234          const exit = yield* Effect.exit(step(dir, fs, git))
235          if (Exit.isFailure(exit)) {
236            yield* Effect.logError("failed to run migration", { index: i, cause: exit.cause })
237            break
238          }
239          yield* fs.writeWithDirs(marker, String(i + 1))
240        }
241        return { dir }
242      }),
243    )
244 
245    const fail = (target: string): Effect.Effect<never, NotFoundError> =>
246      Effect.fail(new NotFoundError({ message: `Resource not found: ${target}` }))
247 
248    const wrap = <A>(target: string, body: Effect.Effect<A, FSUtil.Error>) =>
249      body.pipe(Effect.catchIf(missing, () => fail(target)))
250 
251    const writeJson = Effect.fnUntraced(function* (target: string, content: unknown) {
252      yield* fs.writeWithDirs(target, JSON.stringify(content, null, 2))
253    })
254 
255    const withResolved = <A, E>(
256      key: string[],
257      fn: (target: string, rw: TxReentrantLock.TxReentrantLock) => Effect.Effect<A, E>,
258    ): Effect.Effect<A, E | FSUtil.Error> =>
259      Effect.scoped(
260        Effect.gen(function* () {
261          const target = file((yield* state).dir, key)
262          return yield* fn(target, yield* RcMap.get(locks, target))
263        }),
264      )
265 
266    const remove: Interface["remove"] = Effect.fn("Storage.remove")(function* (key: string[]) {
267      yield* withResolved(key, (target, rw) =>
268        TxReentrantLock.withWriteLock(rw, fs.remove(target).pipe(Effect.catchIf(missing, () => Effect.void))),
269      )
270    })
271 
272    const read: Interface["read"] = <T>(key: string[]) =>
273      Effect.gen(function* () {
274        const value = yield* withResolved(key, (target, rw) =>
275          TxReentrantLock.withReadLock(rw, wrap(target, fs.readJson(target))),
276        )
277        return value as T
278      })
279 
280    const update: Interface["update"] = <T>(key: string[], fn: (draft: T) => void) =>
281      Effect.gen(function* () {
282        const value = yield* withResolved(key, (target, rw) =>
283          TxReentrantLock.withWriteLock(
284            rw,
285            Effect.gen(function* () {
286              const content = yield* wrap(target, fs.readJson(target))
287              fn(content as T)
288              yield* writeJson(target, content)
289              return content
290            }),
291          ),
292        )
293        return value as T
294      })
295 
296    const write: Interface["write"] = (key: string[], content: unknown) =>
297      Effect.gen(function* () {
298        yield* withResolved(key, (target, rw) => TxReentrantLock.withWriteLock(rw, writeJson(target, content)))
299      })
300 
301    const list: Interface["list"] = Effect.fn("Storage.list")(function* (prefix: string[]) {
302      const dir = (yield* state).dir
303      const cwd = path.join(dir, ...prefix)
304      const result = yield* fs
305        .glob("**/*", {
306          cwd,
307          include: "file",
308        })
309        .pipe(Effect.catch(() => Effect.succeed<string[]>([])))
310      return result
311        .map((x) => [...prefix, ...x.slice(0, -5).split(path.sep)])
312        .toSorted((a, b) => a.join("/").localeCompare(b.join("/")))
313    })
314 
315    return Service.of({
316      remove,
317      read,
318      update,
319      write,
320      list,
321    })
322  }),
323)
324 
325export const node = LayerNode.make({ service: Service, layer: layer, deps: [FSUtil.node, Git.node] })
326 
327export * as Storage from "./storage"
328