diff --git a/apps/server/integration/providerService.integration.test.ts b/apps/server/integration/providerService.integration.test.ts index 57e93c5acdd0bb2daeee5b06e5486a1bdf293c23..e703af4b1f45bd03132bfd9e8336c8d7320679a2 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 e97696fbadd3943d76904d4135dd1dcd74ba943e..c655a0f36b6725d0b2b8dd8e186a32399cb83211 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 967766a7a4e3c6692b942a120ab51f5ffa885ae2..130222408a6e25c34c3804b6401f6d59a0173390 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 8de145ca338b861c9ee96edb60fc32a8def0d70b..e1064c279047960e9c74d726429d7e60d6792aa6 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/AuthPairingLinks.ts b/apps/server/src/persistence/AuthPairingLinks.ts new file mode 100644 index 0000000000000000000000000000000000000000..add90f0480318a01f855169b08b509f830035316 --- /dev/null +++ b/apps/server/src/persistence/AuthPairingLinks.ts @@ -0,0 +1,284 @@ +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, +} 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, + { + 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 => + Schema.isSchemaError(cause) + ? toPersistenceDecodeError(decodeOperation)(cause) + : toPersistenceSqlError(sqlOperation)(cause); +} + +export const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const createPairingLinkRow = SqlSchema.void({ + Request: CreateAuthPairingLinkInput, + execute: (input) => + sql` + INSERT INTO auth_pairing_links ( + id, + credential, + method, + scopes, + subject, + label, + proof_key_thumbprint, + created_at, + expires_at, + consumed_at, + revoked_at + ) + VALUES ( + ${input.id}, + ${input.credential}, + ${input.method}, + ${JSON.stringify(input.scopes)}, + ${input.subject}, + ${input.label}, + ${input.proofKeyThumbprint}, + ${input.createdAt}, + ${input.expiresAt}, + NULL, + NULL + ) + `, + }); + + const consumeAvailablePairingLinkRow = SqlSchema.findOneOption({ + Request: ConsumeAuthPairingLinkInput, + Result: AuthPairingLinkRecord, + execute: ({ credential, proofKeyThumbprint, consumedAt, now }) => + sql` + UPDATE auth_pairing_links + SET consumed_at = ${consumedAt} + WHERE credential = ${credential} + AND revoked_at IS NULL + AND consumed_at IS NULL + AND expires_at > ${now} + AND ( + proof_key_thumbprint IS NULL + OR proof_key_thumbprint = ${proofKeyThumbprint} + ) + RETURNING + id AS "id", + credential AS "credential", + method AS "method", + scopes AS "scopes", + subject AS "subject", + label AS "label", + proof_key_thumbprint AS "proofKeyThumbprint", + created_at AS "createdAt", + expires_at AS "expiresAt", + consumed_at AS "consumedAt", + revoked_at AS "revokedAt" + `, + }); + + const listActivePairingLinkRows = SqlSchema.findAll({ + Request: ListActiveAuthPairingLinksInput, + Result: AuthPairingLinkRecord, + execute: ({ now }) => + sql` + SELECT + id AS "id", + credential AS "credential", + method AS "method", + scopes AS "scopes", + subject AS "subject", + label AS "label", + proof_key_thumbprint AS "proofKeyThumbprint", + created_at AS "createdAt", + expires_at AS "expiresAt", + consumed_at AS "consumedAt", + revoked_at AS "revokedAt" + FROM auth_pairing_links + WHERE revoked_at IS NULL + AND consumed_at IS NULL + AND expires_at > ${now} + ORDER BY created_at DESC, id DESC + `, + }); + + const revokePairingLinkRow = SqlSchema.findAll({ + Request: RevokeAuthPairingLinkInput, + Result: Schema.Struct({ id: Schema.String }), + execute: ({ id, revokedAt }) => + sql` + UPDATE auth_pairing_links + SET revoked_at = ${revokedAt} + WHERE id = ${id} + AND revoked_at IS NULL + AND consumed_at IS NULL + RETURNING id AS "id" + `, + }); + + const getPairingLinkRowByCredential = SqlSchema.findOneOption({ + Request: GetAuthPairingLinkByCredentialInput, + Result: AuthPairingLinkRecord, + execute: ({ credential }) => + sql` + SELECT + id AS "id", + credential AS "credential", + method AS "method", + scopes AS "scopes", + subject AS "subject", + label AS "label", + proof_key_thumbprint AS "proofKeyThumbprint", + created_at AS "createdAt", + expires_at AS "expiresAt", + consumed_at AS "consumedAt", + revoked_at AS "revokedAt" + FROM auth_pairing_links + WHERE credential = ${credential} + `, + }); + + const create: AuthPairingLinkRepository["Service"]["create"] = (input) => + createPairingLinkRow(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "AuthPairingLinkRepository.create:query", + "AuthPairingLinkRepository.create:encodeRequest", + ), + ), + ); + + const consumeAvailable: AuthPairingLinkRepository["Service"]["consumeAvailable"] = (input) => + consumeAvailablePairingLinkRow(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "AuthPairingLinkRepository.consumeAvailable:query", + "AuthPairingLinkRepository.consumeAvailable:decodeRow", + ), + ), + ); + + const listActive: AuthPairingLinkRepository["Service"]["listActive"] = (input) => + listActivePairingLinkRows(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "AuthPairingLinkRepository.listActive:query", + "AuthPairingLinkRepository.listActive:decodeRows", + ), + ), + ); + + const revoke: AuthPairingLinkRepository["Service"]["revoke"] = (input) => + revokePairingLinkRow(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "AuthPairingLinkRepository.revoke:query", + "AuthPairingLinkRepository.revoke:decodeRows", + ), + ), + Effect.map((rows) => rows.length > 0), + ); + + const getByCredential: AuthPairingLinkRepository["Service"]["getByCredential"] = (input) => + getPairingLinkRowByCredential(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "AuthPairingLinkRepository.getByCredential:query", + "AuthPairingLinkRepository.getByCredential:decodeRow", + ), + ), + ); + + return { + create, + consumeAvailable, + listActive, + revoke, + getByCredential, + } satisfies AuthPairingLinkRepository["Service"]; +}); + +export const layer = Layer.effect(AuthPairingLinkRepository, make); diff --git a/apps/server/src/persistence/AuthSessions.ts b/apps/server/src/persistence/AuthSessions.ts new file mode 100644 index 0000000000000000000000000000000000000000..e3e8a19f5d0f478ebb8d50dcf044dd5f4a31afcd --- /dev/null +++ b/apps/server/src/persistence/AuthSessions.ts @@ -0,0 +1,361 @@ +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, +} 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, + { + 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, + subject: Schema.String, + scopes: Schema.fromJsonString(AuthEnvironmentScopes), + method: ServerAuthSessionMethod, + clientLabel: Schema.NullOr(Schema.String), + clientIpAddress: Schema.NullOr(Schema.String), + clientUserAgent: Schema.NullOr(Schema.String), + clientDeviceType: Schema.Literals(["desktop", "mobile", "tablet", "bot", "unknown"]), + clientOs: Schema.NullOr(Schema.String), + clientBrowser: Schema.NullOr(Schema.String), + issuedAt: Schema.DateTimeUtcFromString, + expiresAt: Schema.DateTimeUtcFromString, + lastConnectedAt: Schema.NullOr(Schema.DateTimeUtcFromString), + revokedAt: Schema.NullOr(Schema.DateTimeUtcFromString), +}); + +function toAuthSessionRecord(row: typeof AuthSessionDbRow.Type): AuthSessionRecord { + return { + sessionId: row.sessionId, + subject: row.subject, + scopes: row.scopes, + method: row.method, + client: { + label: row.clientLabel, + ipAddress: row.clientIpAddress, + userAgent: row.clientUserAgent, + deviceType: row.clientDeviceType, + os: row.clientOs, + browser: row.clientBrowser, + }, + issuedAt: row.issuedAt, + expiresAt: row.expiresAt, + lastConnectedAt: row.lastConnectedAt, + revokedAt: row.revokedAt, + }; +} + +function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) { + return (cause: unknown): AuthSessionRepositoryError => + Schema.isSchemaError(cause) + ? toPersistenceDecodeError(decodeOperation)(cause) + : toPersistenceSqlError(sqlOperation)(cause); +} + +export const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + const createSessionRow = SqlSchema.void({ + Request: CreateAuthSessionInput, + execute: (input) => + sql` + INSERT INTO auth_sessions ( + session_id, + subject, + scopes, + method, + client_label, + client_ip_address, + client_user_agent, + client_device_type, + client_os, + client_browser, + issued_at, + expires_at, + revoked_at + ) + VALUES ( + ${input.sessionId}, + ${input.subject}, + ${JSON.stringify(input.scopes)}, + ${input.method}, + ${input.client.label}, + ${input.client.ipAddress}, + ${input.client.userAgent}, + ${input.client.deviceType}, + ${input.client.os}, + ${input.client.browser}, + ${input.issuedAt}, + ${input.expiresAt}, + NULL + ) + `, + }); + + const getSessionRowById = SqlSchema.findOneOption({ + Request: GetAuthSessionByIdInput, + Result: AuthSessionDbRow, + execute: ({ sessionId }) => + sql` + SELECT + session_id AS "sessionId", + subject AS "subject", + scopes AS "scopes", + method AS "method", + client_label AS "clientLabel", + client_ip_address AS "clientIpAddress", + client_user_agent AS "clientUserAgent", + client_device_type AS "clientDeviceType", + client_os AS "clientOs", + client_browser AS "clientBrowser", + issued_at AS "issuedAt", + expires_at AS "expiresAt", + last_connected_at AS "lastConnectedAt", + revoked_at AS "revokedAt" + FROM auth_sessions + WHERE session_id = ${sessionId} + `, + }); + + const listActiveSessionRows = SqlSchema.findAll({ + Request: ListActiveAuthSessionsInput, + Result: AuthSessionDbRow, + execute: ({ now }) => + sql` + SELECT + session_id AS "sessionId", + subject AS "subject", + scopes AS "scopes", + method AS "method", + client_label AS "clientLabel", + client_ip_address AS "clientIpAddress", + client_user_agent AS "clientUserAgent", + client_device_type AS "clientDeviceType", + client_os AS "clientOs", + client_browser AS "clientBrowser", + issued_at AS "issuedAt", + expires_at AS "expiresAt", + last_connected_at AS "lastConnectedAt", + revoked_at AS "revokedAt" + FROM auth_sessions + WHERE revoked_at IS NULL + AND expires_at > ${now} + ORDER BY issued_at DESC, session_id DESC + `, + }); + + const setLastConnectedAtRow = SqlSchema.void({ + Request: SetAuthSessionLastConnectedAtInput, + execute: ({ sessionId, lastConnectedAt }) => + sql` + UPDATE auth_sessions + SET last_connected_at = ${lastConnectedAt} + WHERE session_id = ${sessionId} + AND revoked_at IS NULL + `, + }); + + const revokeSessionRows = SqlSchema.findAll({ + Request: RevokeAuthSessionInput, + Result: Schema.Struct({ sessionId: AuthSessionId }), + execute: ({ sessionId, revokedAt }) => + sql` + UPDATE auth_sessions + SET revoked_at = ${revokedAt} + WHERE session_id = ${sessionId} + AND revoked_at IS NULL + RETURNING session_id AS "sessionId" + `, + }); + + const revokeOtherSessionRows = SqlSchema.findAll({ + Request: RevokeOtherAuthSessionsInput, + Result: Schema.Struct({ sessionId: AuthSessionId }), + execute: ({ currentSessionId, revokedAt }) => + sql` + UPDATE auth_sessions + SET revoked_at = ${revokedAt} + WHERE session_id <> ${currentSessionId} + AND revoked_at IS NULL + RETURNING session_id AS "sessionId" + `, + }); + + const create: AuthSessionRepository["Service"]["create"] = (input) => + createSessionRow(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "AuthSessionRepository.create:query", + "AuthSessionRepository.create:encodeRequest", + ), + ), + ); + + const getById: AuthSessionRepository["Service"]["getById"] = (input) => + getSessionRowById(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "AuthSessionRepository.getById:query", + "AuthSessionRepository.getById:decodeRow", + ), + ), + Effect.flatMap((rowOption) => + Option.match(rowOption, { + onNone: () => Effect.succeed(Option.none()), + onSome: (row) => Effect.succeed(Option.some(toAuthSessionRecord(row))), + }), + ), + ); + + const listActive: AuthSessionRepository["Service"]["listActive"] = (input) => + listActiveSessionRows(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "AuthSessionRepository.listActive:query", + "AuthSessionRepository.listActive:decodeRows", + ), + ), + Effect.flatMap((rows) => Effect.succeed(rows.map((row) => toAuthSessionRecord(row)))), + ); + + const revoke: AuthSessionRepository["Service"]["revoke"] = (input) => + revokeSessionRows(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "AuthSessionRepository.revoke:query", + "AuthSessionRepository.revoke:decodeRows", + ), + ), + Effect.map((rows) => rows.length > 0), + ); + + const revokeAllExcept: AuthSessionRepository["Service"]["revokeAllExcept"] = (input) => + revokeOtherSessionRows(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "AuthSessionRepository.revokeAllExcept:query", + "AuthSessionRepository.revokeAllExcept:decodeRows", + ), + ), + Effect.map((rows) => rows.map((row) => row.sessionId)), + ); + + const setLastConnectedAt: AuthSessionRepository["Service"]["setLastConnectedAt"] = (input) => + setLastConnectedAtRow(input).pipe( + Effect.mapError( + toPersistenceSqlOrDecodeError( + "AuthSessionRepository.setLastConnectedAt:query", + "AuthSessionRepository.setLastConnectedAt:encodeRequest", + ), + ), + ); + + return { + create, + getById, + listActive, + revoke, + revokeAllExcept, + setLastConnectedAt, + } satisfies AuthSessionRepository["Service"]; +}); + +export const layer = Layer.effect(AuthSessionRepository, make); diff --git a/apps/server/src/persistence/Layers/AuthPairingLinks.ts b/apps/server/src/persistence/Layers/AuthPairingLinks.ts deleted file mode 100644 index 9d2760d14490b5b4f9d32dc82c7f8fc45a8f7fd3..0000000000000000000000000000000000000000 --- a/apps/server/src/persistence/Layers/AuthPairingLinks.ts +++ /dev/null @@ -1,220 +0,0 @@ -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 Schema from "effect/Schema"; - -import { - toPersistenceDecodeError, - toPersistenceSqlError, - type AuthPairingLinkRepositoryError, -} from "../Errors.ts"; -import { - AuthPairingLinkRecord, - AuthPairingLinkRepository, - type AuthPairingLinkRepositoryShape, - ConsumeAuthPairingLinkInput, - CreateAuthPairingLinkInput, - GetAuthPairingLinkByCredentialInput, - ListActiveAuthPairingLinksInput, - RevokeAuthPairingLinkInput, -} from "../Services/AuthPairingLinks.ts"; - -function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) { - return (cause: unknown): AuthPairingLinkRepositoryError => - Schema.isSchemaError(cause) - ? toPersistenceDecodeError(decodeOperation)(cause) - : toPersistenceSqlError(sqlOperation)(cause); -} - -const makeAuthPairingLinkRepository = Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - - const createPairingLinkRow = SqlSchema.void({ - Request: CreateAuthPairingLinkInput, - execute: (input) => - sql` - INSERT INTO auth_pairing_links ( - id, - credential, - method, - scopes, - subject, - label, - proof_key_thumbprint, - created_at, - expires_at, - consumed_at, - revoked_at - ) - VALUES ( - ${input.id}, - ${input.credential}, - ${input.method}, - ${JSON.stringify(input.scopes)}, - ${input.subject}, - ${input.label}, - ${input.proofKeyThumbprint}, - ${input.createdAt}, - ${input.expiresAt}, - NULL, - NULL - ) - `, - }); - - const consumeAvailablePairingLinkRow = SqlSchema.findOneOption({ - Request: ConsumeAuthPairingLinkInput, - Result: AuthPairingLinkRecord, - execute: ({ credential, proofKeyThumbprint, consumedAt, now }) => - sql` - UPDATE auth_pairing_links - SET consumed_at = ${consumedAt} - WHERE credential = ${credential} - AND revoked_at IS NULL - AND consumed_at IS NULL - AND expires_at > ${now} - AND ( - proof_key_thumbprint IS NULL - OR proof_key_thumbprint = ${proofKeyThumbprint} - ) - RETURNING - id AS "id", - credential AS "credential", - method AS "method", - scopes AS "scopes", - subject AS "subject", - label AS "label", - proof_key_thumbprint AS "proofKeyThumbprint", - created_at AS "createdAt", - expires_at AS "expiresAt", - consumed_at AS "consumedAt", - revoked_at AS "revokedAt" - `, - }); - - const listActivePairingLinkRows = SqlSchema.findAll({ - Request: ListActiveAuthPairingLinksInput, - Result: AuthPairingLinkRecord, - execute: ({ now }) => - sql` - SELECT - id AS "id", - credential AS "credential", - method AS "method", - scopes AS "scopes", - subject AS "subject", - label AS "label", - proof_key_thumbprint AS "proofKeyThumbprint", - created_at AS "createdAt", - expires_at AS "expiresAt", - consumed_at AS "consumedAt", - revoked_at AS "revokedAt" - FROM auth_pairing_links - WHERE revoked_at IS NULL - AND consumed_at IS NULL - AND expires_at > ${now} - ORDER BY created_at DESC, id DESC - `, - }); - - const revokePairingLinkRow = SqlSchema.findAll({ - Request: RevokeAuthPairingLinkInput, - Result: Schema.Struct({ id: Schema.String }), - execute: ({ id, revokedAt }) => - sql` - UPDATE auth_pairing_links - SET revoked_at = ${revokedAt} - WHERE id = ${id} - AND revoked_at IS NULL - AND consumed_at IS NULL - RETURNING id AS "id" - `, - }); - - const getPairingLinkRowByCredential = SqlSchema.findOneOption({ - Request: GetAuthPairingLinkByCredentialInput, - Result: AuthPairingLinkRecord, - execute: ({ credential }) => - sql` - SELECT - id AS "id", - credential AS "credential", - method AS "method", - scopes AS "scopes", - subject AS "subject", - label AS "label", - proof_key_thumbprint AS "proofKeyThumbprint", - created_at AS "createdAt", - expires_at AS "expiresAt", - consumed_at AS "consumedAt", - revoked_at AS "revokedAt" - FROM auth_pairing_links - WHERE credential = ${credential} - `, - }); - - const create: AuthPairingLinkRepositoryShape["create"] = (input) => - createPairingLinkRow(input).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "AuthPairingLinkRepository.create:query", - "AuthPairingLinkRepository.create:encodeRequest", - ), - ), - ); - - const consumeAvailable: AuthPairingLinkRepositoryShape["consumeAvailable"] = (input) => - consumeAvailablePairingLinkRow(input).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "AuthPairingLinkRepository.consumeAvailable:query", - "AuthPairingLinkRepository.consumeAvailable:decodeRow", - ), - ), - ); - - const listActive: AuthPairingLinkRepositoryShape["listActive"] = (input) => - listActivePairingLinkRows(input).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "AuthPairingLinkRepository.listActive:query", - "AuthPairingLinkRepository.listActive:decodeRows", - ), - ), - ); - - const revoke: AuthPairingLinkRepositoryShape["revoke"] = (input) => - revokePairingLinkRow(input).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "AuthPairingLinkRepository.revoke:query", - "AuthPairingLinkRepository.revoke:decodeRows", - ), - ), - Effect.map((rows) => rows.length > 0), - ); - - const getByCredential: AuthPairingLinkRepositoryShape["getByCredential"] = (input) => - getPairingLinkRowByCredential(input).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "AuthPairingLinkRepository.getByCredential:query", - "AuthPairingLinkRepository.getByCredential:decodeRow", - ), - ), - ); - - return { - create, - consumeAvailable, - listActive, - revoke, - getByCredential, - } satisfies AuthPairingLinkRepositoryShape; -}); - -export const AuthPairingLinkRepositoryLive = Layer.effect( - AuthPairingLinkRepository, - makeAuthPairingLinkRepository, -); diff --git a/apps/server/src/persistence/Layers/AuthSessions.ts b/apps/server/src/persistence/Layers/AuthSessions.ts deleted file mode 100644 index ab84e3fa041695a3f552b9b7cb28e0d11561fc83..0000000000000000000000000000000000000000 --- a/apps/server/src/persistence/Layers/AuthSessions.ts +++ /dev/null @@ -1,282 +0,0 @@ -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 Effect from "effect/Effect"; -import * as Layer from "effect/Layer"; -import * as Option from "effect/Option"; -import * as Schema from "effect/Schema"; - -import { - toPersistenceDecodeError, - toPersistenceSqlError, - type AuthSessionRepositoryError, -} from "../Errors.ts"; -import { - AuthSessionRecord, - AuthSessionRepository, - type AuthSessionRepositoryShape, - CreateAuthSessionInput, - GetAuthSessionByIdInput, - ListActiveAuthSessionsInput, - RevokeAuthSessionInput, - RevokeOtherAuthSessionsInput, - SetAuthSessionLastConnectedAtInput, -} from "../Services/AuthSessions.ts"; - -const AuthSessionDbRow = Schema.Struct({ - sessionId: AuthSessionId, - subject: Schema.String, - scopes: Schema.fromJsonString(AuthEnvironmentScopes), - method: ServerAuthSessionMethod, - clientLabel: Schema.NullOr(Schema.String), - clientIpAddress: Schema.NullOr(Schema.String), - clientUserAgent: Schema.NullOr(Schema.String), - clientDeviceType: Schema.Literals(["desktop", "mobile", "tablet", "bot", "unknown"]), - clientOs: Schema.NullOr(Schema.String), - clientBrowser: Schema.NullOr(Schema.String), - issuedAt: Schema.DateTimeUtcFromString, - expiresAt: Schema.DateTimeUtcFromString, - lastConnectedAt: Schema.NullOr(Schema.DateTimeUtcFromString), - revokedAt: Schema.NullOr(Schema.DateTimeUtcFromString), -}); - -function toAuthSessionRecord(row: typeof AuthSessionDbRow.Type): AuthSessionRecord { - return { - sessionId: row.sessionId, - subject: row.subject, - scopes: row.scopes, - method: row.method, - client: { - label: row.clientLabel, - ipAddress: row.clientIpAddress, - userAgent: row.clientUserAgent, - deviceType: row.clientDeviceType, - os: row.clientOs, - browser: row.clientBrowser, - }, - issuedAt: row.issuedAt, - expiresAt: row.expiresAt, - lastConnectedAt: row.lastConnectedAt, - revokedAt: row.revokedAt, - }; -} - -function toPersistenceSqlOrDecodeError(sqlOperation: string, decodeOperation: string) { - return (cause: unknown): AuthSessionRepositoryError => - Schema.isSchemaError(cause) - ? toPersistenceDecodeError(decodeOperation)(cause) - : toPersistenceSqlError(sqlOperation)(cause); -} - -const makeAuthSessionRepository = Effect.gen(function* () { - const sql = yield* SqlClient.SqlClient; - - const createSessionRow = SqlSchema.void({ - Request: CreateAuthSessionInput, - execute: (input) => - sql` - INSERT INTO auth_sessions ( - session_id, - subject, - scopes, - method, - client_label, - client_ip_address, - client_user_agent, - client_device_type, - client_os, - client_browser, - issued_at, - expires_at, - revoked_at - ) - VALUES ( - ${input.sessionId}, - ${input.subject}, - ${JSON.stringify(input.scopes)}, - ${input.method}, - ${input.client.label}, - ${input.client.ipAddress}, - ${input.client.userAgent}, - ${input.client.deviceType}, - ${input.client.os}, - ${input.client.browser}, - ${input.issuedAt}, - ${input.expiresAt}, - NULL - ) - `, - }); - - const getSessionRowById = SqlSchema.findOneOption({ - Request: GetAuthSessionByIdInput, - Result: AuthSessionDbRow, - execute: ({ sessionId }) => - sql` - SELECT - session_id AS "sessionId", - subject AS "subject", - scopes AS "scopes", - method AS "method", - client_label AS "clientLabel", - client_ip_address AS "clientIpAddress", - client_user_agent AS "clientUserAgent", - client_device_type AS "clientDeviceType", - client_os AS "clientOs", - client_browser AS "clientBrowser", - issued_at AS "issuedAt", - expires_at AS "expiresAt", - last_connected_at AS "lastConnectedAt", - revoked_at AS "revokedAt" - FROM auth_sessions - WHERE session_id = ${sessionId} - `, - }); - - const listActiveSessionRows = SqlSchema.findAll({ - Request: ListActiveAuthSessionsInput, - Result: AuthSessionDbRow, - execute: ({ now }) => - sql` - SELECT - session_id AS "sessionId", - subject AS "subject", - scopes AS "scopes", - method AS "method", - client_label AS "clientLabel", - client_ip_address AS "clientIpAddress", - client_user_agent AS "clientUserAgent", - client_device_type AS "clientDeviceType", - client_os AS "clientOs", - client_browser AS "clientBrowser", - issued_at AS "issuedAt", - expires_at AS "expiresAt", - last_connected_at AS "lastConnectedAt", - revoked_at AS "revokedAt" - FROM auth_sessions - WHERE revoked_at IS NULL - AND expires_at > ${now} - ORDER BY issued_at DESC, session_id DESC - `, - }); - - const setLastConnectedAtRow = SqlSchema.void({ - Request: SetAuthSessionLastConnectedAtInput, - execute: ({ sessionId, lastConnectedAt }) => - sql` - UPDATE auth_sessions - SET last_connected_at = ${lastConnectedAt} - WHERE session_id = ${sessionId} - AND revoked_at IS NULL - `, - }); - - const revokeSessionRows = SqlSchema.findAll({ - Request: RevokeAuthSessionInput, - Result: Schema.Struct({ sessionId: AuthSessionId }), - execute: ({ sessionId, revokedAt }) => - sql` - UPDATE auth_sessions - SET revoked_at = ${revokedAt} - WHERE session_id = ${sessionId} - AND revoked_at IS NULL - RETURNING session_id AS "sessionId" - `, - }); - - const revokeOtherSessionRows = SqlSchema.findAll({ - Request: RevokeOtherAuthSessionsInput, - Result: Schema.Struct({ sessionId: AuthSessionId }), - execute: ({ currentSessionId, revokedAt }) => - sql` - UPDATE auth_sessions - SET revoked_at = ${revokedAt} - WHERE session_id <> ${currentSessionId} - AND revoked_at IS NULL - RETURNING session_id AS "sessionId" - `, - }); - - const create: AuthSessionRepositoryShape["create"] = (input) => - createSessionRow(input).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "AuthSessionRepository.create:query", - "AuthSessionRepository.create:encodeRequest", - ), - ), - ); - - const getById: AuthSessionRepositoryShape["getById"] = (input) => - getSessionRowById(input).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "AuthSessionRepository.getById:query", - "AuthSessionRepository.getById:decodeRow", - ), - ), - Effect.flatMap((rowOption) => - Option.match(rowOption, { - onNone: () => Effect.succeed(Option.none()), - onSome: (row) => Effect.succeed(Option.some(toAuthSessionRecord(row))), - }), - ), - ); - - const listActive: AuthSessionRepositoryShape["listActive"] = (input) => - listActiveSessionRows(input).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "AuthSessionRepository.listActive:query", - "AuthSessionRepository.listActive:decodeRows", - ), - ), - Effect.flatMap((rows) => Effect.succeed(rows.map((row) => toAuthSessionRecord(row)))), - ); - - const revoke: AuthSessionRepositoryShape["revoke"] = (input) => - revokeSessionRows(input).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "AuthSessionRepository.revoke:query", - "AuthSessionRepository.revoke:decodeRows", - ), - ), - Effect.map((rows) => rows.length > 0), - ); - - const revokeAllExcept: AuthSessionRepositoryShape["revokeAllExcept"] = (input) => - revokeOtherSessionRows(input).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "AuthSessionRepository.revokeAllExcept:query", - "AuthSessionRepository.revokeAllExcept:decodeRows", - ), - ), - Effect.map((rows) => rows.map((row) => row.sessionId)), - ); - - const setLastConnectedAt: AuthSessionRepositoryShape["setLastConnectedAt"] = (input) => - setLastConnectedAtRow(input).pipe( - Effect.mapError( - toPersistenceSqlOrDecodeError( - "AuthSessionRepository.setLastConnectedAt:query", - "AuthSessionRepository.setLastConnectedAt:encodeRequest", - ), - ), - ); - - return { - create, - getById, - listActive, - revoke, - revokeAllExcept, - setLastConnectedAt, - } satisfies AuthSessionRepositoryShape; -}); - -export const AuthSessionRepositoryLive = Layer.effect( - AuthSessionRepository, - makeAuthSessionRepository, -); diff --git a/apps/server/src/persistence/Layers/ProviderSessionRuntime.ts b/apps/server/src/persistence/Layers/ProviderSessionRuntime.ts index 9ee5c82bb53d39c0830a086ff1258badd4fb0435..52e4f8f74088f0b936815320cb4eb1a1ec369c49 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 6b91b5bd07baab00a250e7ec841c07a13058c8b5..fd49edf0529b4047e13b8bcfb4875f2ccfd3825a 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 0000000000000000000000000000000000000000..6bbbfbd4e19d554aa309fef48a0f934c876ac939 --- /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 c8745982d2913b1410ea3780d2229009ed8e4b92..0000000000000000000000000000000000000000 --- 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 c08956bdd716abb68d08a8529247c775d446de72..0000000000000000000000000000000000000000 --- 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 125f4fa5bbf23f4bc437aba9bdd98a4b7a5d306d..0000000000000000000000000000000000000000 --- 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 6a72bf6994141ec71da4321a0ac7e0961a23e9f6..8581a11213b7cebc7c22934a7c7556a8ebb27357 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 f9793ca9d1fe12ad89280709dcdf3151fbe4c493..c5d60a69a225a600890362362874f9dd7d894654 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 0508f6c8cb34c812a1ea169cc33c0946fc30a599..23075bd9a06e7dee71f9d43ac29d81cdca704f57 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 18e6166c1cddfe5fad741a32c48370a617c05d08..e976c183a4386c50b825fd657ee6d4f121efdd5e 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 373dc61bad8a4745b8fccf1bc603ecb5514bedb4..f1e900c0b5a468305acfb025cf787c594e3e4621 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