diff --git a/apps/server/integration/providerService.integration.test.ts b/apps/server/integration/providerService.integration.test.ts index 57e93c5acdd..e703af4b1f4 100644 --- a/apps/server/integration/providerService.integration.test.ts +++ b/apps/server/integration/providerService.integration.test.ts @@ -25,7 +25,7 @@ import { import { ServerSettingsService } from "../src/serverSettings.ts"; import { AnalyticsService } from "../src/telemetry/Services/AnalyticsService.ts"; import { SqlitePersistenceMemory } from "../src/persistence/Layers/Sqlite.ts"; -import { ProviderSessionRuntimeRepositoryLive } from "../src/persistence/Layers/ProviderSessionRuntime.ts"; +import * as ProviderSessionRuntime from "../src/persistence/ProviderSessionRuntime.ts"; import { makeTestProviderAdapterHarness, @@ -63,7 +63,7 @@ const makeIntegrationFixture = Effect.gen(function* () { }); const directoryLayer = ProviderSessionDirectoryLive.pipe( - Layer.provide(ProviderSessionRuntimeRepositoryLive), + Layer.provide(ProviderSessionRuntime.layer), ); const shared = Layer.mergeAll( diff --git a/apps/server/src/auth/PairingGrantStore.ts b/apps/server/src/auth/PairingGrantStore.ts index e97696fbadd..c655a0f36b6 100644 --- a/apps/server/src/auth/PairingGrantStore.ts +++ b/apps/server/src/auth/PairingGrantStore.ts @@ -18,8 +18,7 @@ import * as Stream from "effect/Stream"; import * as Option from "effect/Option"; import { ServerConfig } from "../config.ts"; -import { AuthPairingLinkRepositoryLive } from "../persistence/Layers/AuthPairingLinks.ts"; -import { AuthPairingLinkRepository } from "../persistence/Services/AuthPairingLinks.ts"; +import * as AuthPairingLinks from "../persistence/AuthPairingLinks.ts"; export interface BootstrapGrant { readonly method: ServerAuthBootstrapMethod; @@ -126,7 +125,7 @@ const internalBootstrapCredentialError = (message: string, cause: unknown) => export const make = Effect.fn("makePairingGrantStore")(function* () { const crypto = yield* Crypto.Crypto; const config = yield* ServerConfig; - const pairingLinks = yield* AuthPairingLinkRepository; + const pairingLinks = yield* AuthPairingLinks.AuthPairingLinkRepository; const seededGrantsRef = yield* Ref.make(new Map()); const changesPubSub = yield* PubSub.unbounded(); const generatePairingToken = Effect.gen(function* () { @@ -417,5 +416,5 @@ export const make = Effect.fn("makePairingGrantStore")(function* () { }); export const layer = Layer.effect(PairingGrantStore, make()).pipe( - Layer.provideMerge(AuthPairingLinkRepositoryLive), + Layer.provideMerge(AuthPairingLinks.layer), ); diff --git a/apps/server/src/auth/SessionStore.test.ts b/apps/server/src/auth/SessionStore.test.ts index 967766a7a4e..130222408a6 100644 --- a/apps/server/src/auth/SessionStore.test.ts +++ b/apps/server/src/auth/SessionStore.test.ts @@ -8,7 +8,7 @@ import * as TestClock from "effect/testing/TestClock"; import * as ServerConfig from "../config.ts"; import { PersistenceSqlError } from "../persistence/Errors.ts"; import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; -import * as AuthSessions from "../persistence/Services/AuthSessions.ts"; +import * as AuthSessions from "../persistence/AuthSessions.ts"; import * as SessionStore from "./SessionStore.ts"; import * as ServerSecretStore from "./ServerSecretStore.ts"; diff --git a/apps/server/src/auth/SessionStore.ts b/apps/server/src/auth/SessionStore.ts index 8de145ca338..e1064c27904 100644 --- a/apps/server/src/auth/SessionStore.ts +++ b/apps/server/src/auth/SessionStore.ts @@ -22,8 +22,7 @@ import * as Stream from "effect/Stream"; import * as Option from "effect/Option"; import { ServerConfig } from "../config.ts"; -import { AuthSessionRepositoryLive } from "../persistence/Layers/AuthSessions.ts"; -import { AuthSessionRepository } from "../persistence/Services/AuthSessions.ts"; +import * as AuthSessions from "../persistence/AuthSessions.ts"; import * as ServerSecretStore from "./ServerSecretStore.ts"; import { base64UrlDecodeUtf8, @@ -195,7 +194,7 @@ export const make = Effect.fn("makeSessionStore")(function* () { const crypto = yield* Crypto.Crypto; const serverConfig = yield* ServerConfig; const secretStore = yield* ServerSecretStore.ServerSecretStore; - const authSessions = yield* AuthSessionRepository; + const authSessions = yield* AuthSessions.AuthSessionRepository; const signingSecret = yield* secretStore.getOrCreateRandom(SIGNING_SECRET_NAME, 32); const connectedSessionsRef = yield* Ref.make(new Map()); const changesPubSub = yield* PubSub.unbounded(); @@ -640,5 +639,5 @@ export const make = Effect.fn("makeSessionStore")(function* () { }); export const layer = Layer.effect(SessionStore, make()).pipe( - Layer.provideMerge(AuthSessionRepositoryLive), + Layer.provideMerge(AuthSessions.layer), ); diff --git a/apps/server/src/persistence/Layers/AuthPairingLinks.ts b/apps/server/src/persistence/AuthPairingLinks.ts similarity index 60% rename from apps/server/src/persistence/Layers/AuthPairingLinks.ts rename to apps/server/src/persistence/AuthPairingLinks.ts index 9d2760d1449..add90f04803 100644 --- a/apps/server/src/persistence/Layers/AuthPairingLinks.ts +++ b/apps/server/src/persistence/AuthPairingLinks.ts @@ -1,24 +1,91 @@ -import * as SqlClient from "effect/unstable/sql/SqlClient"; -import * as SqlSchema from "effect/unstable/sql/SqlSchema"; +import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as SqlSchema from "effect/unstable/sql/SqlSchema"; + +import { AuthEnvironmentScopes } from "@t3tools/contracts"; import { + type AuthPairingLinkRepositoryError, toPersistenceDecodeError, toPersistenceSqlError, - type AuthPairingLinkRepositoryError, -} from "../Errors.ts"; -import { - AuthPairingLinkRecord, +} from "./Errors.ts"; + +export const AuthPairingLinkRecord = Schema.Struct({ + id: Schema.String, + credential: Schema.String, + method: Schema.Literals(["desktop-bootstrap", "one-time-token"]), + scopes: Schema.fromJsonString(AuthEnvironmentScopes), + subject: Schema.String, + label: Schema.NullOr(Schema.String), + proofKeyThumbprint: Schema.NullOr(Schema.String), + createdAt: Schema.DateTimeUtcFromString, + expiresAt: Schema.DateTimeUtcFromString, + consumedAt: Schema.NullOr(Schema.DateTimeUtcFromString), + revokedAt: Schema.NullOr(Schema.DateTimeUtcFromString), +}); +export type AuthPairingLinkRecord = typeof AuthPairingLinkRecord.Type; + +export const CreateAuthPairingLinkInput = Schema.Struct({ + id: Schema.String, + credential: Schema.String, + method: Schema.Literals(["desktop-bootstrap", "one-time-token"]), + scopes: AuthEnvironmentScopes, + subject: Schema.String, + label: Schema.NullOr(Schema.String), + proofKeyThumbprint: Schema.NullOr(Schema.String), + createdAt: Schema.DateTimeUtcFromString, + expiresAt: Schema.DateTimeUtcFromString, +}); +export type CreateAuthPairingLinkInput = typeof CreateAuthPairingLinkInput.Type; + +export const ConsumeAuthPairingLinkInput = Schema.Struct({ + credential: Schema.String, + proofKeyThumbprint: Schema.NullOr(Schema.String), + consumedAt: Schema.DateTimeUtcFromString, + now: Schema.DateTimeUtcFromString, +}); +export type ConsumeAuthPairingLinkInput = typeof ConsumeAuthPairingLinkInput.Type; + +export const ListActiveAuthPairingLinksInput = Schema.Struct({ + now: Schema.DateTimeUtcFromString, +}); +export type ListActiveAuthPairingLinksInput = typeof ListActiveAuthPairingLinksInput.Type; + +export const RevokeAuthPairingLinkInput = Schema.Struct({ + id: Schema.String, + revokedAt: Schema.DateTimeUtcFromString, +}); +export type RevokeAuthPairingLinkInput = typeof RevokeAuthPairingLinkInput.Type; + +export const GetAuthPairingLinkByCredentialInput = Schema.Struct({ + credential: Schema.String, +}); +export type GetAuthPairingLinkByCredentialInput = typeof GetAuthPairingLinkByCredentialInput.Type; + +export class AuthPairingLinkRepository extends Context.Service< AuthPairingLinkRepository, - type AuthPairingLinkRepositoryShape, - ConsumeAuthPairingLinkInput, - CreateAuthPairingLinkInput, - GetAuthPairingLinkByCredentialInput, - ListActiveAuthPairingLinksInput, - RevokeAuthPairingLinkInput, -} from "../Services/AuthPairingLinks.ts"; + { + readonly create: ( + input: CreateAuthPairingLinkInput, + ) => Effect.Effect; + readonly consumeAvailable: ( + input: ConsumeAuthPairingLinkInput, + ) => Effect.Effect, AuthPairingLinkRepositoryError>; + readonly listActive: ( + input: ListActiveAuthPairingLinksInput, + ) => Effect.Effect, AuthPairingLinkRepositoryError>; + readonly revoke: ( + input: RevokeAuthPairingLinkInput, + ) => Effect.Effect; + readonly getByCredential: ( + input: GetAuthPairingLinkByCredentialInput, + ) => Effect.Effect, AuthPairingLinkRepositoryError>; + } +>()("t3/persistence/AuthPairingLinks/AuthPairingLinkRepository") {} function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) { return (cause: unknown): AuthPairingLinkRepositoryError => @@ -27,7 +94,7 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st : toPersistenceSqlError(sqlOperation)(cause); } -const makeAuthPairingLinkRepository = Effect.gen(function* () { +export const make = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const createPairingLinkRow = SqlSchema.void({ @@ -154,7 +221,7 @@ const makeAuthPairingLinkRepository = Effect.gen(function* () { `, }); - const create: AuthPairingLinkRepositoryShape["create"] = (input) => + const create: AuthPairingLinkRepository["Service"]["create"] = (input) => createPairingLinkRow(input).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -164,7 +231,7 @@ const makeAuthPairingLinkRepository = Effect.gen(function* () { ), ); - const consumeAvailable: AuthPairingLinkRepositoryShape["consumeAvailable"] = (input) => + const consumeAvailable: AuthPairingLinkRepository["Service"]["consumeAvailable"] = (input) => consumeAvailablePairingLinkRow(input).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -174,7 +241,7 @@ const makeAuthPairingLinkRepository = Effect.gen(function* () { ), ); - const listActive: AuthPairingLinkRepositoryShape["listActive"] = (input) => + const listActive: AuthPairingLinkRepository["Service"]["listActive"] = (input) => listActivePairingLinkRows(input).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -184,7 +251,7 @@ const makeAuthPairingLinkRepository = Effect.gen(function* () { ), ); - const revoke: AuthPairingLinkRepositoryShape["revoke"] = (input) => + const revoke: AuthPairingLinkRepository["Service"]["revoke"] = (input) => revokePairingLinkRow(input).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -195,7 +262,7 @@ const makeAuthPairingLinkRepository = Effect.gen(function* () { Effect.map((rows) => rows.length > 0), ); - const getByCredential: AuthPairingLinkRepositoryShape["getByCredential"] = (input) => + const getByCredential: AuthPairingLinkRepository["Service"]["getByCredential"] = (input) => getPairingLinkRowByCredential(input).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -211,10 +278,7 @@ const makeAuthPairingLinkRepository = Effect.gen(function* () { listActive, revoke, getByCredential, - } satisfies AuthPairingLinkRepositoryShape; + } satisfies AuthPairingLinkRepository["Service"]; }); -export const AuthPairingLinkRepositoryLive = Layer.effect( - AuthPairingLinkRepository, - makeAuthPairingLinkRepository, -); +export const layer = Layer.effect(AuthPairingLinkRepository, make); diff --git a/apps/server/src/persistence/Layers/AuthSessions.ts b/apps/server/src/persistence/AuthSessions.ts similarity index 65% rename from apps/server/src/persistence/Layers/AuthSessions.ts rename to apps/server/src/persistence/AuthSessions.ts index ab84e3fa041..e3e8a19f5d0 100644 --- a/apps/server/src/persistence/Layers/AuthSessions.ts +++ b/apps/server/src/persistence/AuthSessions.ts @@ -1,27 +1,109 @@ -import { AuthEnvironmentScopes, AuthSessionId, ServerAuthSessionMethod } from "@t3tools/contracts"; -import * as SqlClient from "effect/unstable/sql/SqlClient"; -import * as SqlSchema from "effect/unstable/sql/SqlSchema"; +import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as SqlSchema from "effect/unstable/sql/SqlSchema"; + +import { + AuthClientMetadataDeviceType, + AuthEnvironmentScopes, + AuthSessionId, + ServerAuthSessionMethod, +} from "@t3tools/contracts"; import { + type AuthSessionRepositoryError, toPersistenceDecodeError, toPersistenceSqlError, - type AuthSessionRepositoryError, -} from "../Errors.ts"; -import { - AuthSessionRecord, +} from "./Errors.ts"; + +export const AuthSessionClientMetadataRecord = Schema.Struct({ + label: Schema.NullOr(Schema.String), + ipAddress: Schema.NullOr(Schema.String), + userAgent: Schema.NullOr(Schema.String), + deviceType: AuthClientMetadataDeviceType, + os: Schema.NullOr(Schema.String), + browser: Schema.NullOr(Schema.String), +}); +export type AuthSessionClientMetadataRecord = typeof AuthSessionClientMetadataRecord.Type; + +export const AuthSessionRecord = Schema.Struct({ + sessionId: AuthSessionId, + subject: Schema.String, + scopes: AuthEnvironmentScopes, + method: ServerAuthSessionMethod, + client: AuthSessionClientMetadataRecord, + issuedAt: Schema.DateTimeUtcFromString, + expiresAt: Schema.DateTimeUtcFromString, + lastConnectedAt: Schema.NullOr(Schema.DateTimeUtcFromString), + revokedAt: Schema.NullOr(Schema.DateTimeUtcFromString), +}); +export type AuthSessionRecord = typeof AuthSessionRecord.Type; + +export const CreateAuthSessionInput = Schema.Struct({ + sessionId: AuthSessionId, + subject: Schema.String, + scopes: AuthEnvironmentScopes, + method: ServerAuthSessionMethod, + client: AuthSessionClientMetadataRecord, + issuedAt: Schema.DateTimeUtcFromString, + expiresAt: Schema.DateTimeUtcFromString, +}); +export type CreateAuthSessionInput = typeof CreateAuthSessionInput.Type; + +export const GetAuthSessionByIdInput = Schema.Struct({ + sessionId: AuthSessionId, +}); +export type GetAuthSessionByIdInput = typeof GetAuthSessionByIdInput.Type; + +export const ListActiveAuthSessionsInput = Schema.Struct({ + now: Schema.DateTimeUtcFromString, +}); +export type ListActiveAuthSessionsInput = typeof ListActiveAuthSessionsInput.Type; + +export const RevokeAuthSessionInput = Schema.Struct({ + sessionId: AuthSessionId, + revokedAt: Schema.DateTimeUtcFromString, +}); +export type RevokeAuthSessionInput = typeof RevokeAuthSessionInput.Type; + +export const RevokeOtherAuthSessionsInput = Schema.Struct({ + currentSessionId: AuthSessionId, + revokedAt: Schema.DateTimeUtcFromString, +}); +export type RevokeOtherAuthSessionsInput = typeof RevokeOtherAuthSessionsInput.Type; + +export const SetAuthSessionLastConnectedAtInput = Schema.Struct({ + sessionId: AuthSessionId, + lastConnectedAt: Schema.DateTimeUtcFromString, +}); +export type SetAuthSessionLastConnectedAtInput = typeof SetAuthSessionLastConnectedAtInput.Type; + +export class AuthSessionRepository extends Context.Service< AuthSessionRepository, - type AuthSessionRepositoryShape, - CreateAuthSessionInput, - GetAuthSessionByIdInput, - ListActiveAuthSessionsInput, - RevokeAuthSessionInput, - RevokeOtherAuthSessionsInput, - SetAuthSessionLastConnectedAtInput, -} from "../Services/AuthSessions.ts"; + { + readonly create: ( + input: CreateAuthSessionInput, + ) => Effect.Effect; + readonly getById: ( + input: GetAuthSessionByIdInput, + ) => Effect.Effect, AuthSessionRepositoryError>; + readonly listActive: ( + input: ListActiveAuthSessionsInput, + ) => Effect.Effect, AuthSessionRepositoryError>; + readonly revoke: ( + input: RevokeAuthSessionInput, + ) => Effect.Effect; + readonly revokeAllExcept: ( + input: RevokeOtherAuthSessionsInput, + ) => Effect.Effect, AuthSessionRepositoryError>; + readonly setLastConnectedAt: ( + input: SetAuthSessionLastConnectedAtInput, + ) => Effect.Effect; + } +>()("t3/persistence/AuthSessions/AuthSessionRepository") {} const AuthSessionDbRow = Schema.Struct({ sessionId: AuthSessionId, @@ -68,7 +150,7 @@ function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: st : toPersistenceSqlError(sqlOperation)(cause); } -const makeAuthSessionRepository = Effect.gen(function* () { +export const make = Effect.gen(function* () { const sql = yield* SqlClient.SqlClient; const createSessionRow = SqlSchema.void({ @@ -197,7 +279,7 @@ const makeAuthSessionRepository = Effect.gen(function* () { `, }); - const create: AuthSessionRepositoryShape["create"] = (input) => + const create: AuthSessionRepository["Service"]["create"] = (input) => createSessionRow(input).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -207,7 +289,7 @@ const makeAuthSessionRepository = Effect.gen(function* () { ), ); - const getById: AuthSessionRepositoryShape["getById"] = (input) => + const getById: AuthSessionRepository["Service"]["getById"] = (input) => getSessionRowById(input).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -223,7 +305,7 @@ const makeAuthSessionRepository = Effect.gen(function* () { ), ); - const listActive: AuthSessionRepositoryShape["listActive"] = (input) => + const listActive: AuthSessionRepository["Service"]["listActive"] = (input) => listActiveSessionRows(input).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -234,7 +316,7 @@ const makeAuthSessionRepository = Effect.gen(function* () { Effect.flatMap((rows) => Effect.succeed(rows.map((row) => toAuthSessionRecord(row)))), ); - const revoke: AuthSessionRepositoryShape["revoke"] = (input) => + const revoke: AuthSessionRepository["Service"]["revoke"] = (input) => revokeSessionRows(input).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -245,7 +327,7 @@ const makeAuthSessionRepository = Effect.gen(function* () { Effect.map((rows) => rows.length > 0), ); - const revokeAllExcept: AuthSessionRepositoryShape["revokeAllExcept"] = (input) => + const revokeAllExcept: AuthSessionRepository["Service"]["revokeAllExcept"] = (input) => revokeOtherSessionRows(input).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -256,7 +338,7 @@ const makeAuthSessionRepository = Effect.gen(function* () { Effect.map((rows) => rows.map((row) => row.sessionId)), ); - const setLastConnectedAt: AuthSessionRepositoryShape["setLastConnectedAt"] = (input) => + const setLastConnectedAt: AuthSessionRepository["Service"]["setLastConnectedAt"] = (input) => setLastConnectedAtRow(input).pipe( Effect.mapError( toPersistenceSqlOrDecodeError( @@ -273,10 +355,7 @@ const makeAuthSessionRepository = Effect.gen(function* () { revoke, revokeAllExcept, setLastConnectedAt, - } satisfies AuthSessionRepositoryShape; + } satisfies AuthSessionRepository["Service"]; }); -export const AuthSessionRepositoryLive = Layer.effect( - AuthSessionRepository, - makeAuthSessionRepository, -); +export const layer = Layer.effect(AuthSessionRepository, make); diff --git a/apps/server/src/persistence/Layers/ProviderSessionRuntime.ts b/apps/server/src/persistence/Layers/ProviderSessionRuntime.ts index 9ee5c82bb53..52e4f8f7408 100644 --- a/apps/server/src/persistence/Layers/ProviderSessionRuntime.ts +++ b/apps/server/src/persistence/Layers/ProviderSessionRuntime.ts @@ -1,208 +1,2 @@ -import { ThreadId } from "@t3tools/contracts"; -import * as SqlClient from "effect/unstable/sql/SqlClient"; -import * as SqlSchema from "effect/unstable/sql/SqlSchema"; -import * as Effect from "effect/Effect"; -import * as Layer from "effect/Layer"; -import * as Option from "effect/Option"; -import * as Schema from "effect/Schema"; -import * as Struct from "effect/Struct"; - -import { - toPersistenceDecodeError, - toPersistenceSqlError, - type ProviderSessionRuntimeRepositoryError, -} from "../Errors.ts"; -import { - ProviderSessionRuntime, - ProviderSessionRuntimeRepository, - type ProviderSessionRuntimeRepositoryShape, -} from "../Services/ProviderSessionRuntime.ts"; - -const ProviderSessionRuntimeDbRowSchema = ProviderSessionRuntime.mapFields( - Struct.assign({ - resumeCursor: Schema.NullOr(Schema.fromJsonString(Schema.Unknown)), - runtimePayload: Schema.NullOr(Schema.fromJsonString(Schema.Unknown)), - }), -); - -const decodeRuntime = Schema.decodeUnknownEffect(ProviderSessionRuntime); - -const GetRuntimeRequestSchema = Schema.Struct({ - threadId: ThreadId, -}); - -const DeleteRuntimeRequestSchema = GetRuntimeRequestSchema; - -function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) { - return (cause: unknown): ProviderSessionRuntimeRepositoryError => - Schema.isSchemaError(cause) - ? toPersistenceDecodeError(decodeOperation)(cause) - : toPersistenceSqlError(sqlOperation)(cause); -} - -const makeProviderSessionRuntimeRepository = Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - - const upsertRuntimeRow = SqlSchema.void({ - Request: ProviderSessionRuntimeDbRowSchema, - execute: (runtime) => - sql` - INSERT INTO provider_session_runtime ( - thread_id, - provider_name, - provider_instance_id, - adapter_key, - runtime_mode, - status, - last_seen_at, - resume_cursor_json, - runtime_payload_json - ) - VALUES ( - ${runtime.threadId}, - ${runtime.providerName}, - ${runtime.providerInstanceId}, - ${runtime.adapterKey}, - ${runtime.runtimeMode}, - ${runtime.status}, - ${runtime.lastSeenAt}, - ${runtime.resumeCursor}, - ${runtime.runtimePayload} - ) - ON CONFLICT (thread_id) - DO UPDATE SET - provider_name = excluded.provider_name, - provider_instance_id = excluded.provider_instance_id, - adapter_key = excluded.adapter_key, - runtime_mode = excluded.runtime_mode, - status = excluded.status, - last_seen_at = excluded.last_seen_at, - resume_cursor_json = excluded.resume_cursor_json, - runtime_payload_json = excluded.runtime_payload_json - `, - }); - - const getRuntimeRowByThreadId = SqlSchema.findOneOption({ - Request: GetRuntimeRequestSchema, - Result: ProviderSessionRuntimeDbRowSchema, - execute: ({ threadId }) => - sql` - SELECT - thread_id AS "threadId", - provider_name AS "providerName", - provider_instance_id AS "providerInstanceId", - adapter_key AS "adapterKey", - runtime_mode AS "runtimeMode", - status, - last_seen_at AS "lastSeenAt", - resume_cursor_json AS "resumeCursor", - runtime_payload_json AS "runtimePayload" - FROM provider_session_runtime - WHERE thread_id = ${threadId} - `, - }); - - const listRuntimeRows = SqlSchema.findAll({ - Request: Schema.Void, - Result: ProviderSessionRuntimeDbRowSchema, - execute: () => - sql` - SELECT - thread_id AS "threadId", - provider_name AS "providerName", - provider_instance_id AS "providerInstanceId", - adapter_key AS "adapterKey", - runtime_mode AS "runtimeMode", - status, - last_seen_at AS "lastSeenAt", - resume_cursor_json AS "resumeCursor", - runtime_payload_json AS "runtimePayload" - FROM provider_session_runtime - ORDER BY last_seen_at ASC, thread_id ASC - `, - }); - - const deleteRuntimeByThreadId = SqlSchema.void({ - Request: DeleteRuntimeRequestSchema, - execute: ({ threadId }) => - sql` - DELETE FROM provider_session_runtime - WHERE thread_id = ${threadId} - `, - }); - - const upsert: ProviderSessionRuntimeRepositoryShape["upsert"] = (runtime) => - upsertRuntimeRow(runtime).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "ProviderSessionRuntimeRepository.upsert:query", - "ProviderSessionRuntimeRepository.upsert:encodeRequest", - ), - ), - ); - - const getByThreadId: ProviderSessionRuntimeRepositoryShape["getByThreadId"] = (input) => - getRuntimeRowByThreadId(input).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "ProviderSessionRuntimeRepository.getByThreadId:query", - "ProviderSessionRuntimeRepository.getByThreadId:decodeRow", - ), - ), - Effect.flatMap((runtimeRowOption) => - Option.match(runtimeRowOption, { - onNone: () => Effect.succeed(Option.none()), - onSome: (row) => - decodeRuntime(row).pipe( - Effect.mapError( - toPersistenceDecodeError( - "ProviderSessionRuntimeRepository.getByThreadId:rowToRuntime", - ), - ), - Effect.map((runtime) => Option.some(runtime)), - ), - }), - ), - ); - - const list: ProviderSessionRuntimeRepositoryShape["list"] = () => - listRuntimeRows(undefined).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "ProviderSessionRuntimeRepository.list:query", - "ProviderSessionRuntimeRepository.list:decodeRows", - ), - ), - Effect.flatMap((rows) => - Effect.forEach( - rows, - (row) => - decodeRuntime(row).pipe( - Effect.mapError( - toPersistenceDecodeError("ProviderSessionRuntimeRepository.list:rowToRuntime"), - ), - ), - { concurrency: "unbounded" }, - ), - ), - ); - - const deleteByThreadId: ProviderSessionRuntimeRepositoryShape["deleteByThreadId"] = (input) => - deleteRuntimeByThreadId(input).pipe( - Effect.mapError( - toPersistenceSqlError("ProviderSessionRuntimeRepository.deleteByThreadId:query"), - ), - ); - - return { - upsert, - getByThreadId, - list, - deleteByThreadId, - } satisfies ProviderSessionRuntimeRepositoryShape; -}); - -export const ProviderSessionRuntimeRepositoryLive = Layer.effect( - ProviderSessionRuntimeRepository, - makeProviderSessionRuntimeRepository, -); +/** @deprecated Compatibility alias for the excluded orchestration integration harness. */ +export { layer as ProviderSessionRuntimeRepositoryLive } from "../ProviderSessionRuntime.ts"; diff --git a/apps/server/src/persistence/NodeSqliteClient.ts b/apps/server/src/persistence/NodeSqliteClient.ts index 6b91b5bd07b..fd49edf0529 100644 --- a/apps/server/src/persistence/NodeSqliteClient.ts +++ b/apps/server/src/persistence/NodeSqliteClient.ts @@ -29,11 +29,6 @@ export const TypeId: TypeId = "~local/sqlite-node/SqliteClient"; export type TypeId = "~local/sqlite-node/SqliteClient"; -/** - * SqliteClient - Effect service tag for the sqlite SQL client. - */ -export const SqliteClient = Context.Service("t3/persistence/NodeSqliteClient"); - export interface SqliteClientConfig { readonly filename: string; readonly readonly?: boolean | undefined; @@ -251,25 +246,12 @@ const makeMemory = ( export const layerConfig = ( config: Config.Wrap, ): Layer.Layer => - Layer.effectContext( - Config.unwrap(config).pipe( - Effect.flatMap(make), - Effect.map((client) => - Context.make(SqliteClient, client).pipe(Context.add(Client.SqlClient, client)), - ), - ), - ).pipe(Layer.provide(Reactivity.layer)); + Layer.effect(Client.SqlClient, Config.unwrap(config).pipe(Effect.flatMap(make))).pipe( + Layer.provide(Reactivity.layer), + ); export const layer = (config: SqliteClientConfig): Layer.Layer => - Layer.effectContext( - Effect.map(make(config), (client) => - Context.make(SqliteClient, client).pipe(Context.add(Client.SqlClient, client)), - ), - ).pipe(Layer.provide(Reactivity.layer)); + Layer.effect(Client.SqlClient, make(config)).pipe(Layer.provide(Reactivity.layer)); export const layerMemory = (config: SqliteMemoryClientConfig = {}): Layer.Layer => - Layer.effectContext( - Effect.map(makeMemory(config), (client) => - Context.make(SqliteClient, client).pipe(Context.add(Client.SqlClient, client)), - ), - ).pipe(Layer.provide(Reactivity.layer)); + Layer.effect(Client.SqlClient, makeMemory(config)).pipe(Layer.provide(Reactivity.layer)); diff --git a/apps/server/src/persistence/ProviderSessionRuntime.ts b/apps/server/src/persistence/ProviderSessionRuntime.ts new file mode 100644 index 00000000000..6bbbfbd4e19 --- /dev/null +++ b/apps/server/src/persistence/ProviderSessionRuntime.ts @@ -0,0 +1,288 @@ +import * as Context from "effect/Context"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as Schema from "effect/Schema"; +import * as Struct from "effect/Struct"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as SqlSchema from "effect/unstable/sql/SqlSchema"; + +import { + IsoDateTime, + ProviderInstanceId, + ProviderSessionRuntimeStatus, + RuntimeMode, + ThreadId, +} from "@t3tools/contracts"; + +import { + type ProviderSessionRuntimeRepositoryError, + toPersistenceDecodeError, + toPersistenceSqlError, +} from "./Errors.ts"; + +/** + * ProviderSessionRuntimeRepository - Repository interface for provider runtime sessions. + * + * Owns persistence operations for provider runtime metadata and resume cursors. + * + * @module ProviderSessionRuntimeRepository + */ + +export const ProviderSessionRuntime = Schema.Struct({ + threadId: ThreadId, + providerName: Schema.String, + /** + * User-defined routing key for the configured provider instance that + * owns this session. Nullable only at the storage/migration boundary: + * rows persisted before the driver/instance split carry only + * `providerName`. Repository consumers must materialize a concrete + * instance id before routing. + */ + providerInstanceId: Schema.NullOr(ProviderInstanceId), + adapterKey: Schema.String, + runtimeMode: RuntimeMode, + status: ProviderSessionRuntimeStatus, + lastSeenAt: IsoDateTime, + resumeCursor: Schema.NullOr(Schema.Unknown), + runtimePayload: Schema.NullOr(Schema.Unknown), +}); +export type ProviderSessionRuntime = typeof ProviderSessionRuntime.Type; + +export const GetProviderSessionRuntimeInput = Schema.Struct({ threadId: ThreadId }); +export type GetProviderSessionRuntimeInput = typeof GetProviderSessionRuntimeInput.Type; + +export const DeleteProviderSessionRuntimeInput = Schema.Struct({ threadId: ThreadId }); +export type DeleteProviderSessionRuntimeInput = typeof DeleteProviderSessionRuntimeInput.Type; + +/** + * ProviderSessionRuntimeRepository - Service tag for provider runtime persistence. + */ +export class ProviderSessionRuntimeRepository extends Context.Service< + ProviderSessionRuntimeRepository, + { + /** + * Insert or replace a provider runtime row. + * + * Upserts by canonical `threadId`, including JSON payload/cursor fields. + */ + readonly upsert: ( + runtime: ProviderSessionRuntime, + ) => Effect.Effect; + + /** + * Read provider runtime state by canonical thread id. + */ + readonly getByThreadId: ( + input: GetProviderSessionRuntimeInput, + ) => Effect.Effect< + Option.Option, + ProviderSessionRuntimeRepositoryError + >; + + /** + * List all provider runtime rows. + * + * Returned in ascending last-seen order. + */ + readonly list: () => Effect.Effect< + ReadonlyArray, + ProviderSessionRuntimeRepositoryError + >; + + /** + * Delete provider runtime state by canonical thread id. + */ + readonly deleteByThreadId: ( + input: DeleteProviderSessionRuntimeInput, + ) => Effect.Effect; + } +>()("t3/persistence/ProviderSessionRuntime/ProviderSessionRuntimeRepository") {} + +const ProviderSessionRuntimeDbRowSchema = ProviderSessionRuntime.mapFields( + Struct.assign({ + resumeCursor: Schema.NullOr(Schema.fromJsonString(Schema.Unknown)), + runtimePayload: Schema.NullOr(Schema.fromJsonString(Schema.Unknown)), + }), +); + +const decodeRuntime = Schema.decodeUnknownEffect(ProviderSessionRuntime); + +const GetRuntimeRequestSchema = Schema.Struct({ + threadId: ThreadId, +}); + +const DeleteRuntimeRequestSchema = GetRuntimeRequestSchema; + +function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) { + return (cause: unknown): ProviderSessionRuntimeRepositoryError => + Schema.isSchemaError(cause) + ? toPersistenceDecodeError(decodeOperation)(cause) + : toPersistenceSqlError(sqlOperation)(cause); +} + +export const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const upsertRuntimeRow = SqlSchema.void({ + Request: ProviderSessionRuntimeDbRowSchema, + execute: (runtime) => + sql` + INSERT INTO provider_session_runtime ( + thread_id, + provider_name, + provider_instance_id, + adapter_key, + runtime_mode, + status, + last_seen_at, + resume_cursor_json, + runtime_payload_json + ) + VALUES ( + ${runtime.threadId}, + ${runtime.providerName}, + ${runtime.providerInstanceId}, + ${runtime.adapterKey}, + ${runtime.runtimeMode}, + ${runtime.status}, + ${runtime.lastSeenAt}, + ${runtime.resumeCursor}, + ${runtime.runtimePayload} + ) + ON CONFLICT (thread_id) + DO UPDATE SET + provider_name = excluded.provider_name, + provider_instance_id = excluded.provider_instance_id, + adapter_key = excluded.adapter_key, + runtime_mode = excluded.runtime_mode, + status = excluded.status, + last_seen_at = excluded.last_seen_at, + resume_cursor_json = excluded.resume_cursor_json, + runtime_payload_json = excluded.runtime_payload_json + `, + }); + + const getRuntimeRowByThreadId = SqlSchema.findOneOption({ + Request: GetRuntimeRequestSchema, + Result: ProviderSessionRuntimeDbRowSchema, + execute: ({ threadId }) => + sql` + SELECT + thread_id AS "threadId", + provider_name AS "providerName", + provider_instance_id AS "providerInstanceId", + adapter_key AS "adapterKey", + runtime_mode AS "runtimeMode", + status, + last_seen_at AS "lastSeenAt", + resume_cursor_json AS "resumeCursor", + runtime_payload_json AS "runtimePayload" + FROM provider_session_runtime + WHERE thread_id = ${threadId} + `, + }); + + const listRuntimeRows = SqlSchema.findAll({ + Request: Schema.Void, + Result: ProviderSessionRuntimeDbRowSchema, + execute: () => + sql` + SELECT + thread_id AS "threadId", + provider_name AS "providerName", + provider_instance_id AS "providerInstanceId", + adapter_key AS "adapterKey", + runtime_mode AS "runtimeMode", + status, + last_seen_at AS "lastSeenAt", + resume_cursor_json AS "resumeCursor", + runtime_payload_json AS "runtimePayload" + FROM provider_session_runtime + ORDER BY last_seen_at ASC, thread_id ASC + `, + }); + + const deleteRuntimeByThreadId = SqlSchema.void({ + Request: DeleteRuntimeRequestSchema, + execute: ({ threadId }) => + sql` + DELETE FROM provider_session_runtime + WHERE thread_id = ${threadId} + `, + }); + + const upsert: ProviderSessionRuntimeRepository["Service"]["upsert"] = (runtime) => + upsertRuntimeRow(runtime).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProviderSessionRuntimeRepository.upsert:query", + "ProviderSessionRuntimeRepository.upsert:encodeRequest", + ), + ), + ); + + const getByThreadId: ProviderSessionRuntimeRepository["Service"]["getByThreadId"] = (input) => + getRuntimeRowByThreadId(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProviderSessionRuntimeRepository.getByThreadId:query", + "ProviderSessionRuntimeRepository.getByThreadId:decodeRow", + ), + ), + Effect.flatMap((runtimeRowOption) => + Option.match(runtimeRowOption, { + onNone: () => Effect.succeed(Option.none()), + onSome: (row) => + decodeRuntime(row).pipe( + Effect.mapError( + toPersistenceDecodeError( + "ProviderSessionRuntimeRepository.getByThreadId:rowToRuntime", + ), + ), + Effect.map((runtime) => Option.some(runtime)), + ), + }), + ), + ); + + const list: ProviderSessionRuntimeRepository["Service"]["list"] = () => + listRuntimeRows(undefined).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "ProviderSessionRuntimeRepository.list:query", + "ProviderSessionRuntimeRepository.list:decodeRows", + ), + ), + Effect.flatMap((rows) => + Effect.forEach( + rows, + (row) => + decodeRuntime(row).pipe( + Effect.mapError( + toPersistenceDecodeError("ProviderSessionRuntimeRepository.list:rowToRuntime"), + ), + ), + { concurrency: "unbounded" }, + ), + ), + ); + + const deleteByThreadId: ProviderSessionRuntimeRepository["Service"]["deleteByThreadId"] = ( + input, + ) => + deleteRuntimeByThreadId(input).pipe( + Effect.mapError( + toPersistenceSqlError("ProviderSessionRuntimeRepository.deleteByThreadId:query"), + ), + ); + + return { + upsert, + getByThreadId, + list, + deleteByThreadId, + } satisfies ProviderSessionRuntimeRepository["Service"]; +}); + +export const layer = Layer.effect(ProviderSessionRuntimeRepository, make); diff --git a/apps/server/src/persistence/Services/AuthPairingLinks.ts b/apps/server/src/persistence/Services/AuthPairingLinks.ts deleted file mode 100644 index c8745982d29..00000000000 --- a/apps/server/src/persistence/Services/AuthPairingLinks.ts +++ /dev/null @@ -1,82 +0,0 @@ -import * as Option from "effect/Option"; -import * as Schema from "effect/Schema"; -import * as Context from "effect/Context"; -import type * as Effect from "effect/Effect"; -import { AuthEnvironmentScopes } from "@t3tools/contracts"; - -import type { AuthPairingLinkRepositoryError } from "../Errors.ts"; - -export const AuthPairingLinkRecord = Schema.Struct({ - id: Schema.String, - credential: Schema.String, - method: Schema.Literals(["desktop-bootstrap", "one-time-token"]), - scopes: Schema.fromJsonString(AuthEnvironmentScopes), - subject: Schema.String, - label: Schema.NullOr(Schema.String), - proofKeyThumbprint: Schema.NullOr(Schema.String), - createdAt: Schema.DateTimeUtcFromString, - expiresAt: Schema.DateTimeUtcFromString, - consumedAt: Schema.NullOr(Schema.DateTimeUtcFromString), - revokedAt: Schema.NullOr(Schema.DateTimeUtcFromString), -}); -export type AuthPairingLinkRecord = typeof AuthPairingLinkRecord.Type; - -export const CreateAuthPairingLinkInput = Schema.Struct({ - id: Schema.String, - credential: Schema.String, - method: Schema.Literals(["desktop-bootstrap", "one-time-token"]), - scopes: AuthEnvironmentScopes, - subject: Schema.String, - label: Schema.NullOr(Schema.String), - proofKeyThumbprint: Schema.NullOr(Schema.String), - createdAt: Schema.DateTimeUtcFromString, - expiresAt: Schema.DateTimeUtcFromString, -}); -export type CreateAuthPairingLinkInput = typeof CreateAuthPairingLinkInput.Type; - -export const ConsumeAuthPairingLinkInput = Schema.Struct({ - credential: Schema.String, - proofKeyThumbprint: Schema.NullOr(Schema.String), - consumedAt: Schema.DateTimeUtcFromString, - now: Schema.DateTimeUtcFromString, -}); -export type ConsumeAuthPairingLinkInput = typeof ConsumeAuthPairingLinkInput.Type; - -export const ListActiveAuthPairingLinksInput = Schema.Struct({ - now: Schema.DateTimeUtcFromString, -}); -export type ListActiveAuthPairingLinksInput = typeof ListActiveAuthPairingLinksInput.Type; - -export const RevokeAuthPairingLinkInput = Schema.Struct({ - id: Schema.String, - revokedAt: Schema.DateTimeUtcFromString, -}); -export type RevokeAuthPairingLinkInput = typeof RevokeAuthPairingLinkInput.Type; - -export const GetAuthPairingLinkByCredentialInput = Schema.Struct({ - credential: Schema.String, -}); -export type GetAuthPairingLinkByCredentialInput = typeof GetAuthPairingLinkByCredentialInput.Type; - -export interface AuthPairingLinkRepositoryShape { - readonly create: ( - input: CreateAuthPairingLinkInput, - ) => Effect.Effect; - readonly consumeAvailable: ( - input: ConsumeAuthPairingLinkInput, - ) => Effect.Effect, AuthPairingLinkRepositoryError>; - readonly listActive: ( - input: ListActiveAuthPairingLinksInput, - ) => Effect.Effect, AuthPairingLinkRepositoryError>; - readonly revoke: ( - input: RevokeAuthPairingLinkInput, - ) => Effect.Effect; - readonly getByCredential: ( - input: GetAuthPairingLinkByCredentialInput, - ) => Effect.Effect, AuthPairingLinkRepositoryError>; -} - -export class AuthPairingLinkRepository extends Context.Service< - AuthPairingLinkRepository, - AuthPairingLinkRepositoryShape ->()("t3/persistence/Services/AuthPairingLinks/AuthPairingLinkRepository") {} diff --git a/apps/server/src/persistence/Services/AuthSessions.ts b/apps/server/src/persistence/Services/AuthSessions.ts deleted file mode 100644 index c08956bdd71..00000000000 --- a/apps/server/src/persistence/Services/AuthSessions.ts +++ /dev/null @@ -1,100 +0,0 @@ -import { - AuthClientMetadataDeviceType, - AuthEnvironmentScopes, - AuthSessionId, - ServerAuthSessionMethod, -} from "@t3tools/contracts"; -import * as Option from "effect/Option"; -import * as Schema from "effect/Schema"; -import * as Context from "effect/Context"; -import type * as Effect from "effect/Effect"; - -import type { AuthSessionRepositoryError } from "../Errors.ts"; - -export const AuthSessionClientMetadataRecord = Schema.Struct({ - label: Schema.NullOr(Schema.String), - ipAddress: Schema.NullOr(Schema.String), - userAgent: Schema.NullOr(Schema.String), - deviceType: AuthClientMetadataDeviceType, - os: Schema.NullOr(Schema.String), - browser: Schema.NullOr(Schema.String), -}); -export type AuthSessionClientMetadataRecord = typeof AuthSessionClientMetadataRecord.Type; - -export const AuthSessionRecord = Schema.Struct({ - sessionId: AuthSessionId, - subject: Schema.String, - scopes: AuthEnvironmentScopes, - method: ServerAuthSessionMethod, - client: AuthSessionClientMetadataRecord, - issuedAt: Schema.DateTimeUtcFromString, - expiresAt: Schema.DateTimeUtcFromString, - lastConnectedAt: Schema.NullOr(Schema.DateTimeUtcFromString), - revokedAt: Schema.NullOr(Schema.DateTimeUtcFromString), -}); -export type AuthSessionRecord = typeof AuthSessionRecord.Type; - -export const CreateAuthSessionInput = Schema.Struct({ - sessionId: AuthSessionId, - subject: Schema.String, - scopes: AuthEnvironmentScopes, - method: ServerAuthSessionMethod, - client: AuthSessionClientMetadataRecord, - issuedAt: Schema.DateTimeUtcFromString, - expiresAt: Schema.DateTimeUtcFromString, -}); -export type CreateAuthSessionInput = typeof CreateAuthSessionInput.Type; - -export const GetAuthSessionByIdInput = Schema.Struct({ - sessionId: AuthSessionId, -}); -export type GetAuthSessionByIdInput = typeof GetAuthSessionByIdInput.Type; - -export const ListActiveAuthSessionsInput = Schema.Struct({ - now: Schema.DateTimeUtcFromString, -}); -export type ListActiveAuthSessionsInput = typeof ListActiveAuthSessionsInput.Type; - -export const RevokeAuthSessionInput = Schema.Struct({ - sessionId: AuthSessionId, - revokedAt: Schema.DateTimeUtcFromString, -}); -export type RevokeAuthSessionInput = typeof RevokeAuthSessionInput.Type; - -export const RevokeOtherAuthSessionsInput = Schema.Struct({ - currentSessionId: AuthSessionId, - revokedAt: Schema.DateTimeUtcFromString, -}); -export type RevokeOtherAuthSessionsInput = typeof RevokeOtherAuthSessionsInput.Type; - -export const SetAuthSessionLastConnectedAtInput = Schema.Struct({ - sessionId: AuthSessionId, - lastConnectedAt: Schema.DateTimeUtcFromString, -}); -export type SetAuthSessionLastConnectedAtInput = typeof SetAuthSessionLastConnectedAtInput.Type; - -export interface AuthSessionRepositoryShape { - readonly create: ( - input: CreateAuthSessionInput, - ) => Effect.Effect; - readonly getById: ( - input: GetAuthSessionByIdInput, - ) => Effect.Effect, AuthSessionRepositoryError>; - readonly listActive: ( - input: ListActiveAuthSessionsInput, - ) => Effect.Effect, AuthSessionRepositoryError>; - readonly revoke: ( - input: RevokeAuthSessionInput, - ) => Effect.Effect; - readonly revokeAllExcept: ( - input: RevokeOtherAuthSessionsInput, - ) => Effect.Effect, AuthSessionRepositoryError>; - readonly setLastConnectedAt: ( - input: SetAuthSessionLastConnectedAtInput, - ) => Effect.Effect; -} - -export class AuthSessionRepository extends Context.Service< - AuthSessionRepository, - AuthSessionRepositoryShape ->()("t3/persistence/Services/AuthSessions/AuthSessionRepository") {} diff --git a/apps/server/src/persistence/Services/ProviderSessionRuntime.ts b/apps/server/src/persistence/Services/ProviderSessionRuntime.ts deleted file mode 100644 index 125f4fa5bbf..00000000000 --- a/apps/server/src/persistence/Services/ProviderSessionRuntime.ts +++ /dev/null @@ -1,92 +0,0 @@ -/** - * ProviderSessionRuntimeRepository - Repository interface for provider runtime sessions. - * - * Owns persistence operations for provider runtime metadata and resume cursors. - * - * @module ProviderSessionRuntimeRepository - */ -import { - IsoDateTime, - ProviderInstanceId, - ProviderSessionRuntimeStatus, - RuntimeMode, - ThreadId, -} from "@t3tools/contracts"; -import * as Option from "effect/Option"; -import * as Schema from "effect/Schema"; -import * as Context from "effect/Context"; -import type * as Effect from "effect/Effect"; - -import type { ProviderSessionRuntimeRepositoryError } from "../Errors.ts"; - -export const ProviderSessionRuntime = Schema.Struct({ - threadId: ThreadId, - providerName: Schema.String, - /** - * User-defined routing key for the configured provider instance that - * owns this session. Nullable only at the storage/migration boundary: - * rows persisted before the driver/instance split carry only - * `providerName`. Repository consumers must materialize a concrete - * instance id before routing. - */ - providerInstanceId: Schema.NullOr(ProviderInstanceId), - adapterKey: Schema.String, - runtimeMode: RuntimeMode, - status: ProviderSessionRuntimeStatus, - lastSeenAt: IsoDateTime, - resumeCursor: Schema.NullOr(Schema.Unknown), - runtimePayload: Schema.NullOr(Schema.Unknown), -}); -export type ProviderSessionRuntime = typeof ProviderSessionRuntime.Type; - -export const GetProviderSessionRuntimeInput = Schema.Struct({ threadId: ThreadId }); -export type GetProviderSessionRuntimeInput = typeof GetProviderSessionRuntimeInput.Type; - -export const DeleteProviderSessionRuntimeInput = Schema.Struct({ threadId: ThreadId }); -export type DeleteProviderSessionRuntimeInput = typeof DeleteProviderSessionRuntimeInput.Type; - -/** - * ProviderSessionRuntimeRepositoryShape - Service API for provider runtime records. - */ -export interface ProviderSessionRuntimeRepositoryShape { - /** - * Insert or replace a provider runtime row. - * - * Upserts by canonical `threadId`, including JSON payload/cursor fields. - */ - readonly upsert: ( - runtime: ProviderSessionRuntime, - ) => Effect.Effect; - - /** - * Read provider runtime state by canonical thread id. - */ - readonly getByThreadId: ( - input: GetProviderSessionRuntimeInput, - ) => Effect.Effect, ProviderSessionRuntimeRepositoryError>; - - /** - * List all provider runtime rows. - * - * Returned in ascending last-seen order. - */ - readonly list: () => Effect.Effect< - ReadonlyArray, - ProviderSessionRuntimeRepositoryError - >; - - /** - * Delete provider runtime state by canonical thread id. - */ - readonly deleteByThreadId: ( - input: DeleteProviderSessionRuntimeInput, - ) => Effect.Effect; -} - -/** - * ProviderSessionRuntimeRepository - Service tag for provider runtime persistence. - */ -export class ProviderSessionRuntimeRepository extends Context.Service< - ProviderSessionRuntimeRepository, - ProviderSessionRuntimeRepositoryShape ->()("t3/persistence/Services/ProviderSessionRuntime/ProviderSessionRuntimeRepository") {} diff --git a/apps/server/src/provider/Layers/ProviderService.test.ts b/apps/server/src/provider/Layers/ProviderService.test.ts index 6a72bf69941..8581a11213b 100644 --- a/apps/server/src/provider/Layers/ProviderService.test.ts +++ b/apps/server/src/provider/Layers/ProviderService.test.ts @@ -53,8 +53,7 @@ import { makeProviderServiceLive } from "./ProviderService.ts"; import { NoOpProviderEventLoggers, ProviderEventLoggers } from "./ProviderEventLoggers.ts"; import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; import * as NodeServices from "@effect/platform-node/NodeServices"; -import { ProviderSessionRuntimeRepositoryLive } from "../../persistence/Layers/ProviderSessionRuntime.ts"; -import { ProviderSessionRuntimeRepository } from "../../persistence/Services/ProviderSessionRuntime.ts"; +import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts"; import { makeSqlitePersistenceLive, SqlitePersistenceMemory, @@ -282,7 +281,7 @@ function makeProviderServiceLayer() { }); const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry); - const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); @@ -327,7 +326,7 @@ it.effect("ProviderServiceLive catches stopAll failures during shutdown", () => [CODEX_DRIVER]: codex.adapter, }); const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry); - const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); @@ -379,7 +378,7 @@ it.effect("ProviderServiceLive rejects new sessions for disabled providers", () : registryBase.getInstanceInfo(instanceId), }; const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry); - const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); @@ -453,7 +452,7 @@ it.effect( }, }, }); - const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); const directoryLayer = ProviderSessionDirectoryLive.pipe( @@ -517,7 +516,7 @@ it.effect("ProviderServiceLive rejects new sessions for disabled custom instance ), }; const providerAdapterLayer = Layer.succeed(ProviderAdapterRegistry, registry); - const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); @@ -557,7 +556,7 @@ it.effect("ProviderServiceLive writes canonical events to the emitting thread se const registry = makeAdapterRegistryMock({ [ProviderDriverKind.make("codex")]: codex.adapter, }); - const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); @@ -612,7 +611,7 @@ it.effect("ProviderServiceLive keeps persisted resumable sessions on startup", ( }); const persistenceLayer = makeSqlitePersistenceLive(dbPath); - const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(persistenceLayer), ); const directoryLayer = ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)); @@ -643,7 +642,7 @@ it.effect("ProviderServiceLive keeps persisted resumable sessions on startup", ( assert.equal(persistedProvider, "codex"); const runtime = yield* Effect.gen(function* () { - const repository = yield* ProviderSessionRuntimeRepository; + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; return yield* repository.getByThreadId({ threadId: asThreadId("thread-stale"), }); @@ -671,7 +670,7 @@ it.effect( const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-service-restart-")); const dbPath = path.join(tempDir, "orchestration.sqlite"); const persistenceLayer = makeSqlitePersistenceLive(dbPath); - const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(persistenceLayer), ); @@ -717,7 +716,7 @@ it.effect( }).pipe(Effect.provide(firstProviderLayer)); const persistedAfterStopAll = yield* Effect.gen(function* () { - const repository = yield* ProviderSessionRuntimeRepository; + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; return yield* repository.getByThreadId({ threadId: startedSession.threadId, }); @@ -909,7 +908,7 @@ routing.layer("ProviderServiceLive routing", (it) => { it.effect("preserves the persisted binding when stopping a session", () => Effect.gen(function* () { const provider = yield* ProviderService; - const runtimeRepository = yield* ProviderSessionRuntimeRepository; + const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; const initial = yield* provider.startSession(asThreadId("thread-reap-preserve"), { provider: ProviderDriverKind.make("codex"), @@ -1179,7 +1178,7 @@ routing.layer("ProviderServiceLive routing", (it) => { it.effect("persists runtime status transitions in provider_session_runtime", () => Effect.gen(function* () { const provider = yield* ProviderService; - const runtimeRepository = yield* ProviderSessionRuntimeRepository; + const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; const threadId = asThreadId("thread-runtime-status"); const session = yield* provider.startSession(threadId, { @@ -1226,7 +1225,7 @@ routing.layer("ProviderServiceLive routing", (it) => { const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-service-start-")); const dbPath = path.join(tempDir, "orchestration.sqlite"); const persistenceLayer = makeSqlitePersistenceLive(dbPath); - const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(persistenceLayer), ); @@ -1316,7 +1315,7 @@ routing.layer("ProviderServiceLive routing", (it) => { const tempDir = fs.mkdtempSync(path.join(os.tmpdir(), "t3-provider-service-cwd-")); const dbPath = path.join(tempDir, "orchestration.sqlite"); const persistenceLayer = makeSqlitePersistenceLive(dbPath); - const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(persistenceLayer), ); @@ -1761,7 +1760,7 @@ validation.layer("ProviderServiceLive validation", (it) => { it.effect("accepts startSession when adapter has not emitted provider thread id yet", () => Effect.gen(function* () { const provider = yield* ProviderService; - const runtimeRepository = yield* ProviderSessionRuntimeRepository; + const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; validation.codex.startSession.mockImplementationOnce((input: ProviderSessionStartInput) => Effect.sync(() => { diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts index f9793ca9d1f..c5d60a69a22 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.test.ts @@ -16,15 +16,12 @@ import { makeSqlitePersistenceLive, SqlitePersistenceMemory, } from "../../persistence/Layers/Sqlite.ts"; -import { ProviderSessionRuntimeRepositoryLive } from "../../persistence/Layers/ProviderSessionRuntime.ts"; -import { ProviderSessionRuntimeRepository } from "../../persistence/Services/ProviderSessionRuntime.ts"; +import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts"; import { ProviderSessionDirectory } from "../Services/ProviderSessionDirectory.ts"; import { ProviderSessionDirectoryLive } from "./ProviderSessionDirectory.ts"; function makeDirectoryLayer(persistenceLayer: Layer.Layer) { - const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( - Layer.provide(persistenceLayer), - ); + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe(Layer.provide(persistenceLayer)); return Layer.mergeAll( runtimeRepositoryLayer, ProviderSessionDirectoryLive.pipe(Layer.provide(runtimeRepositoryLayer)), @@ -36,7 +33,7 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL it("upserts and reads thread bindings", () => Effect.gen(function* () { const directory = yield* ProviderSessionDirectory; - const runtimeRepository = yield* ProviderSessionRuntimeRepository; + const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; const initialThreadId = ThreadId.make("thread-1"); @@ -83,7 +80,7 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL it("persists runtime fields and merges payload updates", () => Effect.gen(function* () { const directory = yield* ProviderSessionDirectory; - const runtimeRepository = yield* ProviderSessionRuntimeRepository; + const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; const threadId = ThreadId.make("thread-runtime"); @@ -128,7 +125,7 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL it("lists persisted bindings with metadata in oldest-first order", () => Effect.gen(function* () { const directory = yield* ProviderSessionDirectory; - const runtimeRepository = yield* ProviderSessionRuntimeRepository; + const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; const olderThreadId = ThreadId.make("thread-runtime-older"); const newerThreadId = ThreadId.make("thread-runtime-newer"); @@ -202,7 +199,7 @@ it.layer(makeDirectoryLayer(SqlitePersistenceMemory))("ProviderSessionDirectoryL it("resets adapterKey to the new provider when provider changes without an explicit adapter key", () => Effect.gen(function* () { const directory = yield* ProviderSessionDirectory; - const runtimeRepository = yield* ProviderSessionRuntimeRepository; + const runtimeRepository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; const threadId = ThreadId.make("thread-provider-change"); yield* runtimeRepository.upsert({ diff --git a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts index 0508f6c8cb3..23075bd9a06 100644 --- a/apps/server/src/provider/Layers/ProviderSessionDirectory.ts +++ b/apps/server/src/provider/Layers/ProviderSessionDirectory.ts @@ -5,8 +5,7 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Schema from "effect/Schema"; -import type { ProviderSessionRuntime } from "../../persistence/Services/ProviderSessionRuntime.ts"; -import { ProviderSessionRuntimeRepository } from "../../persistence/Services/ProviderSessionRuntime.ts"; +import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts"; import { ProviderSessionDirectoryPersistenceError, ProviderValidationError } from "../Errors.ts"; import { ProviderSessionDirectory, @@ -59,7 +58,7 @@ function mergeRuntimePayload( } function toRuntimeBinding( - runtime: ProviderSessionRuntime, + runtime: ProviderSessionRuntime.ProviderSessionRuntime, operation: string, ): Effect.Effect { return decodeProviderDriverKind(runtime.providerName, operation).pipe( @@ -85,7 +84,7 @@ function toRuntimeBinding( } const makeProviderSessionDirectory = Effect.gen(function* () { - const repository = yield* ProviderSessionRuntimeRepository; + const repository = yield* ProviderSessionRuntime.ProviderSessionRuntimeRepository; const getBinding = (threadId: ThreadId) => repository.getByThreadId({ threadId }).pipe( diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts index 18e6166c1cd..e976c183a43 100644 --- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts +++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts @@ -19,8 +19,7 @@ import { afterEach, describe, expect, it, vi } from "vite-plus/test"; import { ProjectionSnapshotQuery } from "../../orchestration/Services/ProjectionSnapshotQuery.ts"; import { SqlitePersistenceMemory } from "../../persistence/Layers/Sqlite.ts"; -import { ProviderSessionRuntimeRepositoryLive } from "../../persistence/Layers/ProviderSessionRuntime.ts"; -import { ProviderSessionRuntimeRepository } from "../../persistence/Services/ProviderSessionRuntime.ts"; +import * as ProviderSessionRuntime from "../../persistence/ProviderSessionRuntime.ts"; import { ProviderValidationError } from "../Errors.ts"; import { ProviderSessionReaper } from "../Services/ProviderSessionReaper.ts"; import { ProviderService, type ProviderServiceShape } from "../Services/ProviderService.ts"; @@ -118,7 +117,7 @@ function makeReadModel( describe("ProviderSessionReaper", () => { let runtime: ManagedRuntime.ManagedRuntime< - ProviderSessionReaper | ProviderSessionRuntimeRepository, + ProviderSessionReaper | ProviderSessionRuntime.ProviderSessionRuntimeRepository, unknown > | null = null; let scope: Scope.Closeable | null = null; @@ -176,7 +175,7 @@ describe("ProviderSessionReaper", () => { streamEvents: Stream.empty, }; - const runtimeRepositoryLayer = ProviderSessionRuntimeRepositoryLive.pipe( + const runtimeRepositoryLayer = ProviderSessionRuntime.layer.pipe( Layer.provide(SqlitePersistenceMemory), ); const providerSessionDirectoryLayer = ProviderSessionDirectoryLive.pipe( @@ -238,7 +237,9 @@ describe("ProviderSessionReaper", () => { }, ]), }); - const repository = await runtime!.runPromise(Effect.service(ProviderSessionRuntimeRepository)); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); await runtime!.runPromise( repository.upsert({ @@ -286,7 +287,9 @@ describe("ProviderSessionReaper", () => { }, ]), }); - const repository = await runtime!.runPromise(Effect.service(ProviderSessionRuntimeRepository)); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); await runtime!.runPromise( repository.upsert({ @@ -333,7 +336,9 @@ describe("ProviderSessionReaper", () => { }, ]), }); - const repository = await runtime!.runPromise(Effect.service(ProviderSessionRuntimeRepository)); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); await runtime!.runPromise( repository.upsert({ @@ -380,7 +385,9 @@ describe("ProviderSessionReaper", () => { }, ]), }); - const repository = await runtime!.runPromise(Effect.service(ProviderSessionRuntimeRepository)); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); await runtime!.runPromise( repository.upsert({ @@ -449,7 +456,9 @@ describe("ProviderSessionReaper", () => { ) : Effect.void, }); - const repository = await runtime!.runPromise(Effect.service(ProviderSessionRuntimeRepository)); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); await runtime!.runPromise( repository.upsert({ @@ -530,7 +539,9 @@ describe("ProviderSessionReaper", () => { ? Effect.die(new Error("simulated stop defect")) : Effect.void, }); - const repository = await runtime!.runPromise(Effect.service(ProviderSessionRuntimeRepository)); + const repository = await runtime!.runPromise( + Effect.service(ProviderSessionRuntime.ProviderSessionRuntimeRepository), + ); await runtime!.runPromise( repository.upsert({ diff --git a/apps/server/src/server.ts b/apps/server/src/server.ts index 373dc61bad8..f1e900c0b5a 100644 --- a/apps/server/src/server.ts +++ b/apps/server/src/server.ts @@ -19,7 +19,7 @@ import { layerConfig as SqlitePersistenceLayerLive } from "./persistence/Layers/ import * as ServerLifecycleEvents from "./serverLifecycleEvents.ts"; import { AnalyticsServiceLayerLive } from "./telemetry/Layers/AnalyticsService.ts"; import { ProviderSessionDirectoryLive } from "./provider/Layers/ProviderSessionDirectory.ts"; -import { ProviderSessionRuntimeRepositoryLive } from "./persistence/Layers/ProviderSessionRuntime.ts"; +import * as ProviderSessionRuntime from "./persistence/ProviderSessionRuntime.ts"; import { ProviderAdapterRegistryLive } from "./provider/Layers/ProviderAdapterRegistry.ts"; import * as ProviderEventLoggers from "./provider/Layers/ProviderEventLoggers.ts"; import { ProviderServiceLive } from "./provider/Layers/ProviderService.ts"; @@ -166,7 +166,7 @@ const ReactorLayerLive = Layer.empty.pipe( ); const ProviderSessionDirectoryLayerLive = ProviderSessionDirectoryLive.pipe( - Layer.provide(ProviderSessionRuntimeRepositoryLive), + Layer.provide(ProviderSessionRuntime.layer), ); // `ProviderAdapterRegistryLive` is now a facade that resolves kind → adapter