History

feat: add local persistence stores and boundaries

Implement local SourceStore and LocalStateStore (snapshot + WAL) with tests, and move snapshot schemas/contracts out of shared ports into persistence-local. Also consolidate platform/schema error mapping into shared local helpers to keep Effect error handling consistent and reduce duplication.

Rhys Sullivan committed 8 months ago

Commit 0186aa21 parent 814a18718 files · +1115 −96Browse files

Unified · Split

18 files
  1. v2/PLAN.md +2 −1
  2. v2/packages/schema/README.md +2 −1
  3. v2/packages/schema/src/index.ts +0 −1
  4. v2/packages/schema/src/snapshot.ts +0 −34
  5. v2/packages/state-contracts/README.md +2 −0
  6. v2/packages/state-contracts/src/index.ts +5 −0
  7. v2/packages/state-contracts/src/source-store.ts +33 −0
  8. v2/packages/state-local-file/README.md +4 −2
  9. v2/packages/state-local-file/package.json +1 −1
  10. v2/packages/state-local-file/src/index.ts +21 −0
  11. v2/packages/state-local-file/src/local-state-snapshot.ts +36 −0
  12. v2/packages/state-local-file/src/local-state-store.ts +26 −0
  13. v2/packages/state-local-file/src/persistence-error.ts +92 −0
  14. v2/packages/state-local-file/src/source-store.test.ts +161 −0
  15. v2/packages/state-local-file/src/source-store.ts +211 −0
  16. v2/packages/state-local-file/src/state-store.test.ts +179 −0
  17. v2/packages/state-local-file/src/state-store.ts +308 −0
  18. v2/packages/state-local-file/src/tool-artifact-store.ts +32 −56
v2/PLAN.md+2−1
@@ -164,7 +164,8 @@ Current scaffold includes: · 163 unmodified lines
164 - policy, approval164 - policy, approval
165 - task run, sync state165 - task run, sync state
166 - event envelope166 - event envelope
167- snapshot/event schemas167- event envelope schema
168- local snapshot/WAL schemas in `persistence-local`
168 169
169### Convex ID compatibility170### Convex ID compatibility
170 171
v2/packages/schema/README.md+2−1
@@ -6,4 +6,5 @@ Current scaffold includes: · 5 unmodified lines
6- branded domain IDs (`src/ids.ts`)6- branded domain IDs (`src/ids.ts`)
7- shared enums and primitives (`src/enums.ts`, `src/common.ts`)7- shared enums and primitives (`src/enums.ts`, `src/common.ts`)
8- core model schemas in `src/models/*`8- core model schemas in `src/models/*`
9- local snapshot/event schemas (`src/snapshot.ts`)9- core event envelope schema in `src/models/event-envelope.ts`
10- local snapshot/WAL schemas live in `@executor-v2/persistence-local`
v2/packages/schema/src/index.ts+0−1
@@ -14,4 +14,3 @@ export * from "./models/task-run"; · 13 unmodified lines
14export * from "./models/sync-state";14export * from "./models/sync-state";
15export * from "./models/event-envelope";15export * from "./models/event-envelope";
16 16
17export * from "./snapshot";
v2/packages/schema/src/snapshot.ts+0−34
@@ -1,34 +0,0 @@
1import { Schema } from "effect";
2
3import { SchemaVersionSchema, TimestampMsSchema } from "./common";
4import { EventEnvelopeSchema } from "./models/event-envelope";
5import { ApprovalSchema } from "./models/approval";
6import { CredentialRefSchema } from "./models/credential-ref";
7import { OAuthTokenSchema } from "./models/oauth-token";
8import { PolicySchema } from "./models/policy";
9import { ProfileSchema } from "./models/profile";
10import { SourceSchema } from "./models/source";
11import { SyncStateSchema } from "./models/sync-state";
12import { TaskRunSchema } from "./models/task-run";
13import { ToolArtifactSchema } from "./models/tool-artifact";
14import { WorkspaceSchema } from "./models/workspace";
15
16export const StateSnapshotSchema = Schema.Struct({
17 schemaVersion: SchemaVersionSchema,
18 generatedAt: TimestampMsSchema,
19 profile: ProfileSchema,
20 workspaces: Schema.Array(WorkspaceSchema),
21 sources: Schema.Array(SourceSchema),
22 toolArtifacts: Schema.Array(ToolArtifactSchema),
23 credentials: Schema.Array(CredentialRefSchema),
24 oauthTokens: Schema.Array(OAuthTokenSchema),
25 policies: Schema.Array(PolicySchema),
26 approvals: Schema.Array(ApprovalSchema),
27 taskRuns: Schema.Array(TaskRunSchema),
28 syncStates: Schema.Array(SyncStateSchema),
29});
30
31export const StateEventLogSchema = Schema.Array(EventEnvelopeSchema);
32
33export type StateSnapshot = typeof StateSnapshotSchema.Type;
34export type StateEventLog = typeof StateEventLogSchema.Type;
v2/packages/state-contracts/README.md+2−0
@@ -4,4 +4,6 @@ Persistence port interface scaffold for Executor v2. · 3 unmodified lines
4 4
5Current scaffold includes:5Current scaffold includes:
6- `ProfileStore` in `src/profiles.ts`6- `ProfileStore` in `src/profiles.ts`
7- `SourceStore` in `src/source-store.ts`
8- local-only snapshot/WAL state store lives in `@executor-v2/persistence-local`
7- `ToolArtifactStore` in `src/tool-artifacts.ts`9- `ToolArtifactStore` in `src/tool-artifacts.ts`
v2/packages/state-contracts/src/index.ts+5−0
@@ -1,4 +1,9 @@
1export type { ProfileStore } from "./profiles";1export type { ProfileStore } from "./profiles";
2export {
3 SourceStoreError,
4 SourceStoreService,
5 type SourceStore,
6} from "./source-store";
2export {7export {
3 ToolArtifactStoreError,8 ToolArtifactStoreError,
4 ToolArtifactStoreService,9 ToolArtifactStoreService,
v2/packages/state-contracts/src/source-store.ts+33−0
@@ -0,0 +1,33 @@
1import type { Source, SourceId, WorkspaceId } from "@executor-v2/schema";
2import * as Context from "effect/Context";
3import * as Data from "effect/Data";
4import type * as Effect from "effect/Effect";
5import type * as Option from "effect/Option";
6
7export class SourceStoreError extends Data.TaggedError("SourceStoreError")<{
8 operation: string;
9 filePath: string;
10 message: string;
11 reason: string | null;
12 details: string | null;
13}> {}
14
15export interface SourceStore {
16 getById(
17 workspaceId: WorkspaceId,
18 sourceId: SourceId,
19 ): Effect.Effect<Option.Option<Source>, SourceStoreError>;
20
21 listByWorkspace(workspaceId: WorkspaceId): Effect.Effect<ReadonlyArray<Source>, SourceStoreError>;
22
23 upsert(source: Source): Effect.Effect<void, SourceStoreError>;
24
25 removeById(
26 workspaceId: WorkspaceId,
27 sourceId: SourceId,
28 ): Effect.Effect<boolean, SourceStoreError>;
29}
30
31export class SourceStoreService extends Context.Tag(
32 "@executor-v2/persistence-ports/SourceStoreService",
33)<SourceStoreService, SourceStore>() {}
v2/packages/state-local-file/README.md+4−2
@@ -3,7 +3,9 @@ · 2 unmodified lines
3Local file-backed persistence adapter scaffold for Executor v2.3Local file-backed persistence adapter scaffold for Executor v2.
4 4
5Current scaffold includes:5Current scaffold includes:
6- local file `SourceStore` implementation in `src/source-store.ts`
7- local-only snapshot/WAL schemas (`src/local-state-snapshot.ts`) and `LocalStateStore` contract (`src/local-state-store.ts`), plus file implementation (`src/state-store.ts`) for `snapshot.json` + `events.jsonl`
6- local file `ToolArtifactStore` implementation in `src/tool-artifact-store.ts`8- local file `ToolArtifactStore` implementation in `src/tool-artifact-store.ts`
7- `LocalToolArtifactStoreLive` layer for service-first wiring9- `LocalSourceStoreLive`, `LocalStateStoreLive`, and `LocalToolArtifactStoreLive` layers for service-first wiring
8- Effect Platform `FileSystem` / `Path` integration for file operations10- Effect Platform `FileSystem` / `Path` integration for file operations
9- atomic JSON persistence for extracted tool artifacts11- atomic JSON persistence for sources, snapshot, and tool artifacts
v2/packages/state-local-file/package.json+1−1
@@ -6,7 +6,7 @@ · 5 unmodified lines
6 ".": "./src/index.ts"6 ".": "./src/index.ts"
7 },7 },
8 "scripts": {8 "scripts": {
9 "test": "bunx vitest run --no-watch src/tool-artifact-store.test.ts",9 "test": "bunx vitest run --no-watch src/*.test.ts",
10 "typecheck": "bunx tsc --noEmit -p tsconfig.json"10 "typecheck": "bunx tsc --noEmit -p tsconfig.json"
11 },11 },
12 "dependencies": {12 "dependencies": {
v2/packages/state-local-file/src/index.ts+21−0
@@ -1,3 +1,24 @@
1export {
2 LocalSourceStoreLive,
3 makeLocalSourceStore,
4 type LocalSourceStoreOptions,
5} from "./source-store";
6export {
7 LocalStateEventLogSchema,
8 LocalStateSnapshotSchema,
9 type LocalStateEventLog,
10 type LocalStateSnapshot,
11} from "./local-state-snapshot";
12export {
13 LocalStateStoreError,
14 LocalStateStoreService,
15 type LocalStateStore,
16} from "./local-state-store";
17export {
18 LocalStateStoreLive,
19 makeLocalStateStore,
20 type LocalStateStoreOptions,
21} from "./state-store";
1export {22export {
2 LocalToolArtifactStoreLive,23 LocalToolArtifactStoreLive,
3 makeLocalToolArtifactStore,24 makeLocalToolArtifactStore,
v2/packages/state-local-file/src/local-state-snapshot.ts+36−0
@@ -0,0 +1,36 @@
1import {
2 ApprovalSchema,
3 CredentialRefSchema,
4 EventEnvelopeSchema,
5 OAuthTokenSchema,
6 PolicySchema,
7 ProfileSchema,
8 SchemaVersionSchema,
9 SourceSchema,
10 SyncStateSchema,
11 TaskRunSchema,
12 TimestampMsSchema,
13 ToolArtifactSchema,
14 WorkspaceSchema,
15} from "@executor-v2/schema";
16import { Schema } from "effect";
17
18export const LocalStateSnapshotSchema = Schema.Struct({
19 schemaVersion: SchemaVersionSchema,
20 generatedAt: TimestampMsSchema,
21 profile: ProfileSchema,
22 workspaces: Schema.Array(WorkspaceSchema),
23 sources: Schema.Array(SourceSchema),
24 toolArtifacts: Schema.Array(ToolArtifactSchema),
25 credentials: Schema.Array(CredentialRefSchema),
26 oauthTokens: Schema.Array(OAuthTokenSchema),
27 policies: Schema.Array(PolicySchema),
28 approvals: Schema.Array(ApprovalSchema),
29 taskRuns: Schema.Array(TaskRunSchema),
30 syncStates: Schema.Array(SyncStateSchema),
31});
32
33export const LocalStateEventLogSchema = Schema.Array(EventEnvelopeSchema);
34
35export type LocalStateSnapshot = typeof LocalStateSnapshotSchema.Type;
36export type LocalStateEventLog = typeof LocalStateEventLogSchema.Type;
v2/packages/state-local-file/src/local-state-store.ts+26−0
@@ -0,0 +1,26 @@
1import type { EventEnvelope } from "@executor-v2/schema";
2import * as Context from "effect/Context";
3import * as Data from "effect/Data";
4import type * as Effect from "effect/Effect";
5import type * as Option from "effect/Option";
6
7import type { LocalStateSnapshot } from "./local-state-snapshot";
8
9export class LocalStateStoreError extends Data.TaggedError("LocalStateStoreError")<{
10 operation: string;
11 filePath: string;
12 message: string;
13 reason: string | null;
14 details: string | null;
15}> {}
16
17export interface LocalStateStore {
18 getSnapshot(): Effect.Effect<Option.Option<LocalStateSnapshot>, LocalStateStoreError>;
19 writeSnapshot(snapshot: LocalStateSnapshot): Effect.Effect<void, LocalStateStoreError>;
20 readEvents(): Effect.Effect<ReadonlyArray<EventEnvelope>, LocalStateStoreError>;
21 appendEvents(events: ReadonlyArray<EventEnvelope>): Effect.Effect<void, LocalStateStoreError>;
22}
23
24export class LocalStateStoreService extends Context.Tag(
25 "@executor-v2/persistence-local/LocalStateStoreService",
26)<LocalStateStoreService, LocalStateStore>() {}
v2/packages/state-local-file/src/persistence-error.ts+92−0
@@ -0,0 +1,92 @@
1import * as PlatformError from "@effect/platform/Error";
2import * as Effect from "effect/Effect";
3import { pipe } from "effect/Function";
4import * as ParseResult from "effect/ParseResult";
5
6export type PersistenceErrorData = {
7 operation: string;
8 filePath: string;
9 message: string;
10 reason: string | null;
11 details: string | null;
12};
13
14type PersistenceErrorFactory<E> = (data: PersistenceErrorData) => E;
15
16export const withPlatformPersistenceError =
17 <E>(makeError: PersistenceErrorFactory<E>) =>
18 <A>(operation: string, filePath: string) =>
19 (
20 self: Effect.Effect<A, PlatformError.PlatformError>,
21 ): Effect.Effect<A, E> =>
22 pipe(
23 self,
24 Effect.catchTags({
25 SystemError: (cause) =>
26 Effect.fail(
27 makeError({
28 operation,
29 filePath,
30 message: cause.message,
31 reason: cause.reason,
32 details: cause.description ?? null,
33 }),
34 ),
35 BadArgument: (cause) =>
36 Effect.fail(
37 makeError({
38 operation,
39 filePath,
40 message: cause.message,
41 reason: null,
42 details: cause.description ?? null,
43 }),
44 ),
45 }),
46 );
47
48export const withNotFoundFallback =
49 <A>(fallback: A) =>
50 (
51 self: Effect.Effect<A, PlatformError.PlatformError>,
52 ): Effect.Effect<A, PlatformError.PlatformError> =>
53 pipe(
54 self,
55 Effect.catchTag("SystemError", (cause) =>
56 cause.reason === "NotFound" ? Effect.succeed(fallback) : Effect.fail(cause),
57 ),
58 );
59
60export const toSchemaPersistenceError = <E>(
61 makeError: PersistenceErrorFactory<E>,
62 operation: string,
63 filePath: string,
64 message: string,
65 cause: ParseResult.ParseError,
66 details?: string,
67): E =>
68 makeError({
69 operation,
70 filePath,
71 message,
72 reason: "InvalidData",
73 details:
74 details === undefined
75 ? ParseResult.TreeFormatter.formatErrorSync(cause)
76 : `${details}\n${ParseResult.TreeFormatter.formatErrorSync(cause)}`,
77 });
78
79export const toInvalidDataError = <E>(
80 makeError: PersistenceErrorFactory<E>,
81 operation: string,
82 filePath: string,
83 message: string,
84 details: string,
85): E =>
86 makeError({
87 operation,
88 filePath,
89 message,
90 reason: "InvalidData",
91 details,
92 });
v2/packages/state-local-file/src/source-store.test.ts+161−0
@@ -0,0 +1,161 @@
1import * as FileSystem from "@effect/platform/FileSystem";
2import * as Path from "@effect/platform/Path";
3import * as BunContext from "@effect/platform-bun/BunContext";
4import { describe, expect, it } from "@effect/vitest";
5import { Effect, Option, Schema } from "effect";
6
7import {
8 type Source,
9 SourceIdSchema,
10 SourceSchema,
11 WorkspaceIdSchema,
12} from "@executor-v2/schema";
13
14import { makeLocalSourceStore } from "./source-store";
15
16const decodeSource = Schema.decodeUnknownSync(SourceSchema);
17const decodeWorkspaceId = Schema.decodeUnknownSync(WorkspaceIdSchema);
18const decodeSourceId = Schema.decodeUnknownSync(SourceIdSchema);
19
20const wsLocal = decodeWorkspaceId("ws_local");
21const wsRemote = decodeWorkspaceId("ws_remote");
22const srcLocal = decodeSourceId("src_local");
23
24describe("makeLocalSourceStore", () => {
25 it.effect("returns empty values when source file does not exist", () =>
26 Effect.gen(function* () {
27 const fileSystem = yield* FileSystem.FileSystem;
28 const tempDir = yield* fileSystem.makeTempDirectory({
29 prefix: "executor-v2-source-store-empty-",
30 });
31
32 const sourceStore = yield* makeLocalSourceStore({ rootDir: tempDir });
33 const listed = yield* sourceStore.listByWorkspace(wsLocal);
34 expect(listed).toHaveLength(0);
35
36 const fetched = yield* sourceStore.getById(wsLocal, srcLocal);
37 expect(Option.isNone(fetched)).toBe(true);
38 }).pipe(Effect.provide(BunContext.layer)),
39 );
40
41 it.effect("upserts, lists, and removes sources by workspace", () =>
42 Effect.gen(function* () {
43 const fileSystem = yield* FileSystem.FileSystem;
44 const path = yield* Path.Path;
45 const tempDir = yield* fileSystem.makeTempDirectory({
46 prefix: "executor-v2-source-store-roundtrip-",
47 });
48
49 const sourceStore = yield* makeLocalSourceStore({ rootDir: tempDir });
50
51 const now = Date.now();
52 const localSource: Source = decodeSource({
53 id: "src_local",
54 workspaceId: "ws_local",
55 name: "Local API",
56 kind: "openapi",
57 endpoint: "https://openapi.vercel.sh",
58 status: "connected",
59 enabled: true,
60 configJson: "{}",
61 sourceHash: null,
62 lastError: null,
63 createdAt: now,
64 updatedAt: now,
65 });
66 const otherWorkspaceSource: Source = decodeSource({
67 id: "src_remote",
68 workspaceId: "ws_remote",
69 name: "Remote API",
70 kind: "mcp",
71 endpoint: "https://example.com/mcp",
72 status: "draft",
73 enabled: false,
74 configJson: "{}",
75 sourceHash: null,
76 lastError: null,
77 createdAt: now,
78 updatedAt: now,
79 });
80
81 yield* sourceStore.upsert(localSource);
82 yield* sourceStore.upsert(otherWorkspaceSource);
83
84 const listedLocal = yield* sourceStore.listByWorkspace(wsLocal);
85 expect(listedLocal).toHaveLength(1);
86 expect(listedLocal[0]?.id).toBe("src_local");
87
88 const listedRemote = yield* sourceStore.listByWorkspace(wsRemote);
89 expect(listedRemote).toHaveLength(1);
90 expect(listedRemote[0]?.id).toBe("src_remote");
91
92 const fetched = yield* sourceStore.getById(wsLocal, srcLocal);
93 expect(Option.isSome(fetched)).toBe(true);
94 if (Option.isSome(fetched)) {
95 expect(fetched.value.name).toBe("Local API");
96 }
97
98 const removed = yield* sourceStore.removeById(wsLocal, srcLocal);
99 expect(removed).toBe(true);
100
101 const listedAfterRemove = yield* sourceStore.listByWorkspace(wsLocal);
102 expect(listedAfterRemove).toHaveLength(0);
103
104 const removedAgain = yield* sourceStore.removeById(wsLocal, srcLocal);
105 expect(removedAgain).toBe(false);
106
107 const sourcesPath = path.resolve(tempDir, "sources.json");
108 expect(yield* fileSystem.exists(sourcesPath)).toBe(true);
109 }).pipe(Effect.provide(BunContext.layer)),
110 );
111
112 it.effect("upsert replaces an existing source by workspace/source key", () =>
113 Effect.gen(function* () {
114 const fileSystem = yield* FileSystem.FileSystem;
115 const tempDir = yield* fileSystem.makeTempDirectory({
116 prefix: "executor-v2-source-store-upsert-",
117 });
118
119 const sourceStore = yield* makeLocalSourceStore({ rootDir: tempDir });
120
121 const now = Date.now();
122 const first: Source = decodeSource({
123 id: "src_local",
124 workspaceId: "ws_local",
125 name: "Local API",
126 kind: "openapi",
127 endpoint: "https://openapi.vercel.sh",
128 status: "draft",
129 enabled: true,
130 configJson: "{}",
131 sourceHash: null,
132 lastError: null,
133 createdAt: now,
134 updatedAt: now,
135 });
136 const second: Source = decodeSource({
137 id: "src_local",
138 workspaceId: "ws_local",
139 name: "Local API Updated",
140 kind: "openapi",
141 endpoint: "https://openapi.vercel.sh",
142 status: "connected",
143 enabled: false,
144 configJson: '{"team":"acme"}',
145 sourceHash: "hash_1",
146 lastError: null,
147 createdAt: now,
148 updatedAt: now + 1,
149 });
150
151 yield* sourceStore.upsert(first);
152 yield* sourceStore.upsert(second);
153
154 const listed = yield* sourceStore.listByWorkspace(wsLocal);
155 expect(listed).toHaveLength(1);
156 expect(listed[0]?.name).toBe("Local API Updated");
157 expect(listed[0]?.enabled).toBe(false);
158 expect(listed[0]?.sourceHash).toBe("hash_1");
159 }).pipe(Effect.provide(BunContext.layer)),
160 );
161});
v2/packages/state-local-file/src/source-store.ts+211−0
@@ -0,0 +1,211 @@
1import * as FileSystem from "@effect/platform/FileSystem";
2import * as Path from "@effect/platform/Path";
3import {
4 SourceStoreError,
5 SourceStoreService,
6 type SourceStore,
7} from "@executor-v2/persistence-ports";
8import {
9 SourceSchema,
10 type Source,
11 type SourceId,
12 type WorkspaceId,
13} from "@executor-v2/schema";
14import * as Effect from "effect/Effect";
15import { pipe } from "effect/Function";
16import * as Layer from "effect/Layer";
17import * as Option from "effect/Option";
18import type * as ParseResult from "effect/ParseResult";
19import * as Schema from "effect/Schema";
20import * as STM from "effect/STM";
21import * as TSemaphore from "effect/TSemaphore";
22
23import {
24 type PersistenceErrorData,
25 toSchemaPersistenceError,
26 withNotFoundFallback,
27 withPlatformPersistenceError,
28} from "./persistence-error";
29
30const SourceListSchema = Schema.Array(SourceSchema);
31const SourceListFromJsonSchema = Schema.parseJson(SourceListSchema);
32const decodeSourcesFromJson = Schema.decodeUnknown(SourceListFromJsonSchema);
33const encodeSourcesToJson = Schema.encode(SourceListFromJsonSchema);
34
35export type LocalSourceStoreOptions = {
36 rootDir: string;
37 fileName?: string;
38};
39
40const defaultSourcesFilePath = (path: Path.Path, options: LocalSourceStoreOptions): string =>
41 path.resolve(options.rootDir, options.fileName ?? "sources.json");
42
43const makeSourceStoreError = (data: PersistenceErrorData): SourceStoreError =>
44 new SourceStoreError(data);
45
46const mapPlatformSourceStoreError = withPlatformPersistenceError(makeSourceStoreError);
47
48const toSchemaSourceStoreError = (
49 operation: string,
50 filePath: string,
51 cause: ParseResult.ParseError,
52): SourceStoreError =>
53 toSchemaPersistenceError(
54 makeSourceStoreError,
55 operation,
56 filePath,
57 "Invalid persisted source payload",
58 cause,
59 );
60
61const sourceStoreKey = (source: Source): string => `${source.workspaceId}:${source.id}`;
62
63const dedupeSources = (sources: ReadonlyArray<Source>): Array<Source> => {
64 const byKey = new Map<string, Source>();
65 for (const source of sources) {
66 byKey.set(sourceStoreKey(source), source);
67 }
68 return Array.from(byKey.values());
69};
70
71const sortSources = (sources: ReadonlyArray<Source>): Array<Source> =>
72 [...sources].sort((left, right) => {
73 const leftName = left.name.toLowerCase();
74 const rightName = right.name.toLowerCase();
75
76 if (leftName === rightName) {
77 return left.id.localeCompare(right.id);
78 }
79
80 return leftName.localeCompare(rightName);
81 });
82
83const readSources = (
84 fileSystem: FileSystem.FileSystem,
85 filePath: string,
86): Effect.Effect<Array<Source>, SourceStoreError> =>
87 pipe(
88 fileSystem.readFileString(filePath),
89 withNotFoundFallback("[]"),
90 mapPlatformSourceStoreError("read", filePath),
91 Effect.flatMap((rawJson) =>
92 pipe(
93 decodeSourcesFromJson(rawJson.trim().length === 0 ? "[]" : rawJson),
94 Effect.map((sources) => sortSources(dedupeSources(Array.from(sources)))),
95 Effect.mapError((cause) => toSchemaSourceStoreError("decode", filePath, cause)),
96 ),
97 ),
98 );
99
100const writeSources = (
101 fileSystem: FileSystem.FileSystem,
102 path: Path.Path,
103 filePath: string,
104 sources: ReadonlyArray<Source>,
105): Effect.Effect<void, SourceStoreError> => {
106 const directoryPath = path.dirname(filePath);
107
108 return pipe(
109 fileSystem.makeDirectory(directoryPath, { recursive: true }),
110 mapPlatformSourceStoreError("mkdir", directoryPath),
111 Effect.flatMap(() =>
112 pipe(
113 encodeSourcesToJson(sources),
114 Effect.mapError((cause) => toSchemaSourceStoreError("encode", filePath, cause)),
115 ),
116 ),
117 Effect.flatMap((payload) =>
118 pipe(
119 fileSystem.makeTempFile({
120 directory: directoryPath,
121 prefix: `${path.basename(filePath)}.tmp-`,
122 }),
123 mapPlatformSourceStoreError("makeTempFile", directoryPath),
124 Effect.flatMap((tempPath) =>
125 pipe(
126 fileSystem.writeFileString(tempPath, payload),
127 mapPlatformSourceStoreError("write", tempPath),
128 Effect.flatMap(() =>
129 pipe(
130 fileSystem.rename(tempPath, filePath),
131 mapPlatformSourceStoreError("rename", filePath),
132 ),
133 ),
134 ),
135 ),
136 ),
137 ),
138 );
139};
140
141export const makeLocalSourceStore = (
142 options: LocalSourceStoreOptions,
143): Effect.Effect<SourceStore, never, FileSystem.FileSystem | Path.Path> =>
144 Effect.gen(function* () {
145 const fileSystem = yield* FileSystem.FileSystem;
146 const path = yield* Path.Path;
147 const writeSemaphore = yield* STM.commit(TSemaphore.make(1));
148 const filePath = defaultSourcesFilePath(path, options);
149
150 return {
151 getById: (workspaceId: WorkspaceId, sourceId: SourceId) =>
152 pipe(
153 readSources(fileSystem, filePath),
154 Effect.map((sources) =>
155 Option.fromNullable(
156 sources.find(
157 (source) => source.workspaceId === workspaceId && source.id === sourceId,
158 ),
159 ),
160 ),
161 ),
162
163 listByWorkspace: (workspaceId: WorkspaceId) =>
164 pipe(
165 readSources(fileSystem, filePath),
166 Effect.map((sources) =>
167 sources.filter((source) => source.workspaceId === workspaceId),
168 ),
169 ),
170
171 upsert: (source: Source) =>
172 pipe(
173 readSources(fileSystem, filePath),
174 Effect.map((sources) => {
175 const byKey = new Map<string, Source>(
176 sources.map((currentSource) => [sourceStoreKey(currentSource), currentSource]),
177 );
178 byKey.set(sourceStoreKey(source), source);
179 return sortSources(Array.from(byKey.values()));
180 }),
181 Effect.flatMap((nextSources) => writeSources(fileSystem, path, filePath, nextSources)),
182 TSemaphore.withPermit(writeSemaphore),
183 ),
184
185 removeById: (workspaceId: WorkspaceId, sourceId: SourceId) =>
186 pipe(
187 readSources(fileSystem, filePath),
188 Effect.flatMap((sources) => {
189 const nextSources = sources.filter(
190 (source) => !(source.workspaceId === workspaceId && source.id === sourceId),
191 );
192 const removed = nextSources.length !== sources.length;
193
194 if (!removed) {
195 return Effect.succeed(false);
196 }
197
198 return pipe(
199 writeSources(fileSystem, path, filePath, nextSources),
200 Effect.as(true),
201 );
202 }),
203 TSemaphore.withPermit(writeSemaphore),
204 ),
205 };
206 });
207
208export const LocalSourceStoreLive = (
209 options: LocalSourceStoreOptions,
210): Layer.Layer<SourceStoreService, never, FileSystem.FileSystem | Path.Path> =>
211 Layer.effect(SourceStoreService, makeLocalSourceStore(options));
v2/packages/state-local-file/src/state-store.test.ts+179−0
@@ -0,0 +1,179 @@
1import * as FileSystem from "@effect/platform/FileSystem";
2import * as Path from "@effect/platform/Path";
3import * as BunContext from "@effect/platform-bun/BunContext";
4import { describe, expect, it } from "@effect/vitest";
5import { Either, Effect, Option, Schema } from "effect";
6
7import { type EventEnvelope, EventEnvelopeSchema } from "@executor-v2/schema";
8
9import {
10 type LocalStateSnapshot,
11 LocalStateSnapshotSchema,
12} from "./local-state-snapshot";
13import { makeLocalStateStore } from "./state-store";
14
15const decodeSnapshot = Schema.decodeUnknownSync(LocalStateSnapshotSchema);
16const decodeEvent = Schema.decodeUnknownSync(EventEnvelopeSchema);
17
18describe("makeLocalStateStore", () => {
19 it.effect("returns empty state when snapshot and event files do not exist", () =>
20 Effect.gen(function* () {
21 const fileSystem = yield* FileSystem.FileSystem;
22 const tempDir = yield* fileSystem.makeTempDirectory({
23 prefix: "executor-v2-state-store-empty-",
24 });
25
26 const stateStore = yield* makeLocalStateStore({ rootDir: tempDir });
27
28 const snapshot = yield* stateStore.getSnapshot();
29 expect(Option.isNone(snapshot)).toBe(true);
30
31 const events = yield* stateStore.readEvents();
32 expect(events).toHaveLength(0);
33 }).pipe(Effect.provide(BunContext.layer)),
34 );
35
36 it.effect("writes snapshot and appends events", () =>
37 Effect.gen(function* () {
38 const fileSystem = yield* FileSystem.FileSystem;
39 const path = yield* Path.Path;
40 const tempDir = yield* fileSystem.makeTempDirectory({
41 prefix: "executor-v2-state-store-roundtrip-",
42 });
43
44 const stateStore = yield* makeLocalStateStore({ rootDir: tempDir });
45
46 const now = Date.now();
47 const snapshot: LocalStateSnapshot = decodeSnapshot({
48 schemaVersion: 1,
49 generatedAt: now,
50 profile: {
51 id: "prof_local",
52 defaultWorkspaceId: "ws_local",
53 displayName: "Local Profile",
54 runtimeMode: "local",
55 createdAt: now,
56 updatedAt: now,
57 },
58 workspaces: [
59 {
60 id: "ws_local",
61 profileId: "prof_local",
62 name: "Local Workspace",
63 createdAt: now,
64 updatedAt: now,
65 },
66 ],
67 sources: [],
68 toolArtifacts: [],
69 credentials: [],
70 oauthTokens: [],
71 policies: [],
72 approvals: [],
73 taskRuns: [],
74 syncStates: [],
75 });
76
77 const firstEvent: EventEnvelope = decodeEvent({
78 id: "evt_1",
79 workspaceId: "ws_local",
80 sequence: 1,
81 schemaVersion: 1,
82 eventType: "source.added",
83 payloadJson: '{"sourceId":"src_1"}',
84 createdAt: now,
85 });
86 const secondEvent: EventEnvelope = decodeEvent({
87 id: "evt_2",
88 workspaceId: "ws_local",
89 sequence: 2,
90 schemaVersion: 1,
91 eventType: "source.connected",
92 payloadJson: '{"sourceId":"src_1"}',
93 createdAt: now + 1,
94 });
95
96 yield* stateStore.writeSnapshot(snapshot);
97 yield* stateStore.appendEvents([firstEvent]);
98 yield* stateStore.appendEvents([secondEvent]);
99
100 const persistedSnapshot = yield* stateStore.getSnapshot();
101 expect(Option.isSome(persistedSnapshot)).toBe(true);
102 if (Option.isSome(persistedSnapshot)) {
103 expect(persistedSnapshot.value.profile.displayName).toBe("Local Profile");
104 }
105
106 const persistedEvents = yield* stateStore.readEvents();
107 expect(persistedEvents).toHaveLength(2);
108 expect(persistedEvents[0]?.id).toBe("evt_1");
109 expect(persistedEvents[1]?.id).toBe("evt_2");
110
111 const snapshotPath = path.resolve(tempDir, "snapshot.json");
112 const eventsPath = path.resolve(tempDir, "events.jsonl");
113
114 expect(yield* fileSystem.exists(snapshotPath)).toBe(true);
115 expect(yield* fileSystem.exists(eventsPath)).toBe(true);
116 }).pipe(Effect.provide(BunContext.layer)),
117 );
118
119 it.effect("rejects duplicate event ids and duplicate workspace sequence pairs", () =>
120 Effect.gen(function* () {
121 const fileSystem = yield* FileSystem.FileSystem;
122 const tempDir = yield* fileSystem.makeTempDirectory({
123 prefix: "executor-v2-state-store-duplicates-",
124 });
125
126 const stateStore = yield* makeLocalStateStore({ rootDir: tempDir });
127
128 const baseEvent: EventEnvelope = decodeEvent({
129 id: "evt_1",
130 workspaceId: "ws_local",
131 sequence: 1,
132 schemaVersion: 1,
133 eventType: "source.added",
134 payloadJson: '{"sourceId":"src_1"}',
135 createdAt: Date.now(),
136 });
137
138 const duplicateIdEvent: EventEnvelope = decodeEvent({
139 id: "evt_1",
140 workspaceId: "ws_local",
141 sequence: 2,
142 schemaVersion: 1,
143 eventType: "source.updated",
144 payloadJson: '{"sourceId":"src_1"}',
145 createdAt: Date.now() + 1,
146 });
147
148 const duplicateSequenceEvent: EventEnvelope = decodeEvent({
149 id: "evt_2",
150 workspaceId: "ws_local",
151 sequence: 1,
152 schemaVersion: 1,
153 eventType: "source.updated",
154 payloadJson: '{"sourceId":"src_1"}',
155 createdAt: Date.now() + 2,
156 });
157
158 yield* stateStore.appendEvents([baseEvent]);
159
160 const duplicateIdResult = yield* Effect.either(
161 stateStore.appendEvents([duplicateIdEvent]),
162 );
163 expect(Either.isLeft(duplicateIdResult)).toBe(true);
164 if (Either.isLeft(duplicateIdResult)) {
165 expect(duplicateIdResult.left._tag).toBe("LocalStateStoreError");
166 expect(duplicateIdResult.left.operation).toBe("validate_events");
167 }
168
169 const duplicateSequenceResult = yield* Effect.either(
170 stateStore.appendEvents([duplicateSequenceEvent]),
171 );
172 expect(Either.isLeft(duplicateSequenceResult)).toBe(true);
173 if (Either.isLeft(duplicateSequenceResult)) {
174 expect(duplicateSequenceResult.left._tag).toBe("LocalStateStoreError");
175 expect(duplicateSequenceResult.left.operation).toBe("validate_events");
176 }
177 }).pipe(Effect.provide(BunContext.layer)),
178 );
179});
v2/packages/state-local-file/src/state-store.ts+308−0
@@ -0,0 +1,308 @@
1import * as FileSystem from "@effect/platform/FileSystem";
2import * as Path from "@effect/platform/Path";
3import {
4 LocalStateStoreError,
5 LocalStateStoreService,
6 type LocalStateStore,
7} from "./local-state-store";
8import { EventEnvelopeSchema, type EventEnvelope } from "@executor-v2/schema";
9import {
10 LocalStateSnapshotSchema,
11 type LocalStateSnapshot,
12} from "./local-state-snapshot";
13import * as Effect from "effect/Effect";
14import { pipe } from "effect/Function";
15import * as Layer from "effect/Layer";
16import * as Option from "effect/Option";
17import type * as ParseResult from "effect/ParseResult";
18import * as Schema from "effect/Schema";
19import * as STM from "effect/STM";
20import * as TSemaphore from "effect/TSemaphore";
21
22import {
23 type PersistenceErrorData,
24 toInvalidDataError,
25 toSchemaPersistenceError,
26 withNotFoundFallback,
27 withPlatformPersistenceError,
28} from "./persistence-error";
29
30const SnapshotFromJsonSchema = Schema.parseJson(LocalStateSnapshotSchema);
31const decodeSnapshotFromJson = Schema.decodeUnknown(SnapshotFromJsonSchema);
32const encodeSnapshotToJson = Schema.encode(SnapshotFromJsonSchema);
33
34const EventEnvelopeFromJsonSchema = Schema.parseJson(EventEnvelopeSchema);
35const decodeEventFromJson = Schema.decodeUnknown(EventEnvelopeFromJsonSchema);
36const encodeEventToJson = Schema.encode(EventEnvelopeFromJsonSchema);
37
38export type LocalStateStoreOptions = {
39 rootDir: string;
40 snapshotFileName?: string;
41 eventLogFileName?: string;
42};
43
44const defaultSnapshotFilePath = (
45 path: Path.Path,
46 options: LocalStateStoreOptions,
47): string => path.resolve(options.rootDir, options.snapshotFileName ?? "snapshot.json");
48
49const defaultEventLogFilePath = (
50 path: Path.Path,
51 options: LocalStateStoreOptions,
52): string => path.resolve(options.rootDir, options.eventLogFileName ?? "events.jsonl");
53
54const makeLocalStateStoreError = (
55 data: PersistenceErrorData,
56): LocalStateStoreError => new LocalStateStoreError(data);
57
58const mapPlatformLocalStateStoreError = withPlatformPersistenceError(
59 makeLocalStateStoreError,
60);
61
62const toSchemaLocalStateStoreError = (
63 operation: string,
64 filePath: string,
65 cause: ParseResult.ParseError,
66 details?: string,
67): LocalStateStoreError =>
68 toSchemaPersistenceError(
69 makeLocalStateStoreError,
70 operation,
71 filePath,
72 "Invalid persisted state payload",
73 cause,
74 details,
75 );
76
77const toValidationLocalStateStoreError = (
78 operation: string,
79 filePath: string,
80 details: string,
81): LocalStateStoreError =>
82 toInvalidDataError(
83 makeLocalStateStoreError,
84 operation,
85 filePath,
86 "Invalid event log",
87 details,
88 );
89
90const writeAtomicFile = (
91 fileSystem: FileSystem.FileSystem,
92 path: Path.Path,
93 filePath: string,
94 payload: string,
95): Effect.Effect<void, LocalStateStoreError> => {
96 const directoryPath = path.dirname(filePath);
97
98 return pipe(
99 fileSystem.makeDirectory(directoryPath, { recursive: true }),
100 mapPlatformLocalStateStoreError("mkdir", directoryPath),
101 Effect.flatMap(() =>
102 pipe(
103 fileSystem.makeTempFile({
104 directory: directoryPath,
105 prefix: `${path.basename(filePath)}.tmp-`,
106 }),
107 mapPlatformLocalStateStoreError("makeTempFile", directoryPath),
108 Effect.flatMap((tempPath) =>
109 pipe(
110 fileSystem.writeFileString(tempPath, payload),
111 mapPlatformLocalStateStoreError("write", tempPath),
112 Effect.flatMap(() =>
113 pipe(
114 fileSystem.rename(tempPath, filePath),
115 mapPlatformLocalStateStoreError("rename", filePath),
116 ),
117 ),
118 ),
119 ),
120 ),
121 ),
122 );
123};
124
125const validateEventLog = (
126 eventLogFilePath: string,
127 events: ReadonlyArray<EventEnvelope>,
128): Effect.Effect<void, LocalStateStoreError> =>
129 Effect.gen(function* () {
130 const seenEventIds = new Set<string>();
131 const seenWorkspaceSequences = new Set<string>();
132
133 for (const event of events) {
134 if (seenEventIds.has(event.id)) {
135 return yield* toValidationLocalStateStoreError(
136 "validate_events",
137 eventLogFilePath,
138 `Duplicate event id detected: ${event.id}`,
139 );
140 }
141 seenEventIds.add(event.id);
142
143 const workspaceSequenceKey = `${event.workspaceId}:${event.sequence}`;
144 if (seenWorkspaceSequences.has(workspaceSequenceKey)) {
145 return yield* toValidationLocalStateStoreError(
146 "validate_events",
147 eventLogFilePath,
148 `Duplicate sequence for workspace detected: ${workspaceSequenceKey}`,
149 );
150 }
151 seenWorkspaceSequences.add(workspaceSequenceKey);
152 }
153 });
154
155const readSnapshot = (
156 fileSystem: FileSystem.FileSystem,
157 snapshotFilePath: string,
158): Effect.Effect<Option.Option<LocalStateSnapshot>, LocalStateStoreError> =>
159 pipe(
160 fileSystem.readFileString(snapshotFilePath),
161 withNotFoundFallback(""),
162 mapPlatformLocalStateStoreError("read_snapshot", snapshotFilePath),
163 Effect.flatMap((rawJson) => {
164 const trimmed = rawJson.trim();
165 if (trimmed.length === 0) {
166 return Effect.succeed(Option.none<LocalStateSnapshot>());
167 }
168
169 return pipe(
170 decodeSnapshotFromJson(trimmed),
171 Effect.map(Option.some),
172 Effect.mapError((cause) =>
173 toSchemaLocalStateStoreError("decode_snapshot", snapshotFilePath, cause),
174 ),
175 );
176 }),
177 );
178
179const writeSnapshot = (
180 fileSystem: FileSystem.FileSystem,
181 path: Path.Path,
182 snapshotFilePath: string,
183 snapshot: LocalStateSnapshot,
184): Effect.Effect<void, LocalStateStoreError> =>
185 pipe(
186 encodeSnapshotToJson(snapshot),
187 Effect.mapError((cause) =>
188 toSchemaLocalStateStoreError("encode_snapshot", snapshotFilePath, cause),
189 ),
190 Effect.flatMap((snapshotJson) =>
191 writeAtomicFile(fileSystem, path, snapshotFilePath, snapshotJson),
192 ),
193 );
194
195const readEventLog = (
196 fileSystem: FileSystem.FileSystem,
197 eventLogFilePath: string,
198): Effect.Effect<Array<EventEnvelope>, LocalStateStoreError> =>
199 Effect.gen(function* () {
200 const rawContent = yield* pipe(
201 fileSystem.readFileString(eventLogFilePath),
202 withNotFoundFallback(""),
203 mapPlatformLocalStateStoreError("read_events", eventLogFilePath),
204 );
205
206 const lines = rawContent
207 .split(/\r?\n/)
208 .map((line) => line.trim())
209 .filter((line) => line.length > 0);
210
211 if (lines.length === 0) {
212 return [];
213 }
214
215 const events: Array<EventEnvelope> = [];
216
217 for (let index = 0; index < lines.length; index += 1) {
218 const line = lines[index];
219 const event = yield* pipe(
220 decodeEventFromJson(line),
221 Effect.mapError((cause) =>
222 toSchemaLocalStateStoreError(
223 "decode_event",
224 eventLogFilePath,
225 cause,
226 `line ${index + 1}`,
227 ),
228 ),
229 );
230 events.push(event);
231 }
232
233 yield* validateEventLog(eventLogFilePath, events);
234
235 return events;
236 });
237
238const writeEventLog = (
239 fileSystem: FileSystem.FileSystem,
240 path: Path.Path,
241 eventLogFilePath: string,
242 events: ReadonlyArray<EventEnvelope>,
243): Effect.Effect<void, LocalStateStoreError> =>
244 Effect.gen(function* () {
245 const encodedLines: Array<string> = [];
246
247 for (let index = 0; index < events.length; index += 1) {
248 const encodedLine = yield* pipe(
249 encodeEventToJson(events[index]),
250 Effect.mapError((cause) =>
251 toSchemaLocalStateStoreError(
252 "encode_event",
253 eventLogFilePath,
254 cause,
255 `event index ${index}`,
256 ),
257 ),
258 );
259 encodedLines.push(encodedLine);
260 }
261
262 const payload = encodedLines.length === 0 ? "" : `${encodedLines.join("\n")}\n`;
263
264 yield* writeAtomicFile(fileSystem, path, eventLogFilePath, payload);
265 });
266
267export const makeLocalStateStore = (
268 options: LocalStateStoreOptions,
269): Effect.Effect<LocalStateStore, never, FileSystem.FileSystem | Path.Path> =>
270 Effect.gen(function* () {
271 const fileSystem = yield* FileSystem.FileSystem;
272 const path = yield* Path.Path;
273 const writeSemaphore = yield* STM.commit(TSemaphore.make(1));
274 const snapshotFilePath = defaultSnapshotFilePath(path, options);
275 const eventLogFilePath = defaultEventLogFilePath(path, options);
276
277 return {
278 getSnapshot: () => readSnapshot(fileSystem, snapshotFilePath),
279
280 writeSnapshot: (snapshot) =>
281 pipe(
282 writeSnapshot(fileSystem, path, snapshotFilePath, snapshot),
283 TSemaphore.withPermit(writeSemaphore),
284 ),
285
286 readEvents: () => readEventLog(fileSystem, eventLogFilePath),
287
288 appendEvents: (events) =>
289 pipe(
290 readEventLog(fileSystem, eventLogFilePath),
291 Effect.flatMap((existingEvents) => {
292 const nextEvents = [...existingEvents, ...events];
293 return pipe(
294 validateEventLog(eventLogFilePath, nextEvents),
295 Effect.flatMap(() =>
296 writeEventLog(fileSystem, path, eventLogFilePath, nextEvents),
297 ),
298 );
299 }),
300 TSemaphore.withPermit(writeSemaphore),
301 ),
302 };
303 });
304
305export const LocalStateStoreLive = (
306 options: LocalStateStoreOptions,
307): Layer.Layer<LocalStateStoreService, never, FileSystem.FileSystem | Path.Path> =>
308 Layer.effect(LocalStateStoreService, makeLocalStateStore(options));
v2/packages/state-local-file/src/tool-artifact-store.ts+32−56
@@ -1,5 +1,4 @@
1import * as FileSystem from "@effect/platform/FileSystem";1import * as FileSystem from "@effect/platform/FileSystem";
2import * as PlatformError from "@effect/platform/Error";
3import * as Path from "@effect/platform/Path";2import * as Path from "@effect/platform/Path";
4import {3import {
5 ToolArtifactStoreError,4 ToolArtifactStoreError,
@@ -15,12 +14,19 @@ import { · 9 unmodified lines
15import * as Effect from "effect/Effect";14import * as Effect from "effect/Effect";
16import { pipe } from "effect/Function";15import { pipe } from "effect/Function";
17import * as Option from "effect/Option";16import * as Option from "effect/Option";
18import * as ParseResult from "effect/ParseResult";17import type * as ParseResult from "effect/ParseResult";
19import * as Schema from "effect/Schema";18import * as Schema from "effect/Schema";
20import * as Layer from "effect/Layer";19import * as Layer from "effect/Layer";
21import * as STM from "effect/STM";20import * as STM from "effect/STM";
22import * as TSemaphore from "effect/TSemaphore";21import * as TSemaphore from "effect/TSemaphore";
23 22
23import {
24 type PersistenceErrorData,
25 toSchemaPersistenceError,
26 withNotFoundFallback,
27 withPlatformPersistenceError,
28} from "./persistence-error";
29
24const ToolArtifactListSchema = Schema.Array(ToolArtifactSchema);30const ToolArtifactListSchema = Schema.Array(ToolArtifactSchema);
25const ToolArtifactListFromJsonSchema = Schema.parseJson(ToolArtifactListSchema);31const ToolArtifactListFromJsonSchema = Schema.parseJson(ToolArtifactListSchema);
26const decodeArtifactsFromJson = Schema.decodeUnknown(ToolArtifactListFromJsonSchema);32const decodeArtifactsFromJson = Schema.decodeUnknown(ToolArtifactListFromJsonSchema);
@@ -33,58 +39,26 @@ export type LocalToolArtifactStoreOptions = { · 6 unmodified lines
33const defaultArtifactsFilePath = (path: Path.Path, rootDir: string): string =>39const defaultArtifactsFilePath = (path: Path.Path, rootDir: string): string =>
34 path.resolve(rootDir, "tool-artifacts.json");40 path.resolve(rootDir, "tool-artifacts.json");
35 41
36const toSystemPersistenceError = (42const makeToolArtifactStoreError = (
37 operation: string,43 data: PersistenceErrorData,
38 filePath: string,44): ToolArtifactStoreError => new ToolArtifactStoreError(data);
39 cause: PlatformError.SystemError,
40): ToolArtifactStoreError =>
41 new ToolArtifactStoreError({
42 operation,
43 filePath,
44 message: cause.message,
45 reason: cause.reason,
46 details: cause.description ?? null,
47 });
48 45
49const toBadArgumentPersistenceError = (46const mapPlatformToolArtifactStoreError = withPlatformPersistenceError(
50 operation: string,47 makeToolArtifactStoreError,
51 filePath: string,48);
52 cause: PlatformError.BadArgument,
53): ToolArtifactStoreError =>
54 new ToolArtifactStoreError({
55 operation,
56 filePath,
57 message: cause.message,
58 reason: null,
59 details: cause.description ?? null,
60 });
61 49
62const withPlatformPersistenceError =50const toSchemaToolArtifactStoreError = (
63 <A>(operation: string, filePath: string) =>
64 (
65 self: Effect.Effect<A, PlatformError.PlatformError>,
66 ): Effect.Effect<A, ToolArtifactStoreError> =>
67 pipe(
68 self,
69 Effect.catchTags({
70 SystemError: (cause) =>
71 Effect.fail(toSystemPersistenceError(operation, filePath, cause)),
72 BadArgument: (cause) =>
73 Effect.fail(toBadArgumentPersistenceError(operation, filePath, cause)),
74 }),
75 );
76const toSchemaPersistenceError = (
77 operation: string,51 operation: string,
78 filePath: string,52 filePath: string,
79 cause: ParseResult.ParseError,53 cause: ParseResult.ParseError,
80): ToolArtifactStoreError =>54): ToolArtifactStoreError =>
81 new ToolArtifactStoreError({55 toSchemaPersistenceError(
56 makeToolArtifactStoreError,
82 operation,57 operation,
83 filePath,58 filePath,
84 message: "Invalid persisted tool artifact payload",59 "Invalid persisted tool artifact payload",
85 reason: "InvalidData",60 cause,
86 details: ParseResult.TreeFormatter.formatErrorSync(cause),61 );
87 });
88 62
89const artifactStoreKey = (artifact: ToolArtifact): string =>63const artifactStoreKey = (artifact: ToolArtifact): string =>
90 `${artifact.workspaceId}:${artifact.sourceId}`;64 `${artifact.workspaceId}:${artifact.sourceId}`;
@@ -103,15 +77,15 @@ const readArtifacts = ( · 12 unmodified lines
103): Effect.Effect<Array<ToolArtifact>, ToolArtifactStoreError> =>77): Effect.Effect<Array<ToolArtifact>, ToolArtifactStoreError> =>
104 pipe(78 pipe(
105 fileSystem.readFileString(filePath),79 fileSystem.readFileString(filePath),
106 Effect.catchTag("SystemError", (cause) =>80 withNotFoundFallback("[]"),
107 cause.reason === "NotFound" ? Effect.succeed("[]") : Effect.fail(cause),81 mapPlatformToolArtifactStoreError("read", filePath),
108 ),
109 withPlatformPersistenceError("read", filePath),
110 Effect.flatMap((rawJson) =>82 Effect.flatMap((rawJson) =>
111 pipe(83 pipe(
112 decodeArtifactsFromJson(rawJson.trim().length === 0 ? "[]" : rawJson),84 decodeArtifactsFromJson(rawJson.trim().length === 0 ? "[]" : rawJson),
113 Effect.map((artifacts) => dedupeArtifacts(Array.from(artifacts))),85 Effect.map((artifacts) => dedupeArtifacts(Array.from(artifacts))),
114 Effect.mapError((cause) => toSchemaPersistenceError("decode", filePath, cause)),86 Effect.mapError((cause) =>
87 toSchemaToolArtifactStoreError("decode", filePath, cause),
88 ),
115 ),89 ),
116 ),90 ),
117 );91 );
@@ -126,11 +100,13 @@ const writeArtifacts = ( · 8 unmodified lines
126 100
127 return pipe(101 return pipe(
128 fileSystem.makeDirectory(directoryPath, { recursive: true }),102 fileSystem.makeDirectory(directoryPath, { recursive: true }),
129 withPlatformPersistenceError("mkdir", directoryPath),103 mapPlatformToolArtifactStoreError("mkdir", directoryPath),
130 Effect.flatMap(() =>104 Effect.flatMap(() =>
131 pipe(105 pipe(
132 encodeArtifactsToJson(artifacts),106 encodeArtifactsToJson(artifacts),
133 Effect.mapError((cause) => toSchemaPersistenceError("encode", filePath, cause)),107 Effect.mapError((cause) =>
108 toSchemaToolArtifactStoreError("encode", filePath, cause),
109 ),
134 ),110 ),
135 ),111 ),
136 Effect.flatMap((payload) =>112 Effect.flatMap((payload) =>
@@ -139,15 +115,15 @@ const writeArtifacts = ( · 2 unmodified lines
139 directory: directoryPath,115 directory: directoryPath,
140 prefix: `${path.basename(filePath)}.tmp-`,116 prefix: `${path.basename(filePath)}.tmp-`,
141 }),117 }),
142 withPlatformPersistenceError("makeTempFile", directoryPath),118 mapPlatformToolArtifactStoreError("makeTempFile", directoryPath),
143 Effect.flatMap((tempPath) =>119 Effect.flatMap((tempPath) =>
144 pipe(120 pipe(
145 fileSystem.writeFileString(tempPath, payload),121 fileSystem.writeFileString(tempPath, payload),
146 withPlatformPersistenceError("write", tempPath),122 mapPlatformToolArtifactStoreError("write", tempPath),
147 Effect.flatMap(() =>123 Effect.flatMap(() =>
148 pipe(124 pipe(
149 fileSystem.rename(tempPath, filePath),125 fileSystem.rename(tempPath, filePath),
150 withPlatformPersistenceError("rename", filePath),126 mapPlatformToolArtifactStoreError("rename", filePath),
151 ),127 ),
152 ),128 ),
153 ),129 ),

Keyboard shortcuts

General

Show keyboard shortcuts
?
Focus the current search
/

Issues and pull requests

Open filters
f

Issues, pull requests and Inbox

Select next item
j↓
Select previous item
k↑
Open selected item
Enter
Dismiss filters or clear selection
Esc

Repository code

Go to file
t