import * as Context from "effect/Context";import * as DateTime from "effect/DateTime";import * as Effect from "effect/Effect";import * as Layer from "effect/Layer";import * as Option from "effect/Option";import { and, eq, isNull } from "drizzle-orm";import * as Crypto from "effect/Crypto";import * as Schema from "effect/Schema"; import * as RelayDb from "../db.ts";import { relayDeliveryAttempts } from "../persistence/schema.ts"; export class DeliveryAttemptRecordPersistenceError extends Schema.TaggedErrorClass<DeliveryAttemptRecordPersistenceError>()(  "DeliveryAttemptRecordPersistenceError",  {    operation: Schema.Literals(["record", "claim-source-job", "complete-source-job"]),    sourceJobId: Schema.NullOr(Schema.String),    userId: Schema.NullOr(Schema.String),    environmentId: Schema.NullOr(Schema.String),    threadId: Schema.NullOr(Schema.String),    deviceId: Schema.NullOr(Schema.String),    kind: Schema.NullOr(Schema.String),    cause: Schema.Defect(),  },) {  override get message(): string {    return `Failed to persist APNs delivery attempt during ${this.operation}.`;  }} export interface DeliveryAttemptInput {  readonly userId: string | null;  readonly environmentId: string | null;  readonly threadId: string | null;  readonly deviceId: string | null;  readonly kind: string;  readonly sourceJobId?: string | null;  readonly token: string | null;  readonly apnsStatus?: number;  readonly apnsReason?: string;  readonly apnsId?: string | null;  readonly transportError?: string;} export interface DeliveryAttemptCompletionInput {  readonly sourceJobId: string;  readonly apnsStatus?: number;  readonly apnsReason?: string;  readonly apnsId?: string | null;  readonly transportError?: string;} export type DeliverySourceJobClaimResult = "claimed" | "completed" | "in_flight"; export class DeliveryAttempts extends Context.Service<  DeliveryAttempts,  {    readonly record: (      input: DeliveryAttemptInput,    ) => Effect.Effect<void, DeliveryAttemptRecordPersistenceError>;    readonly claimSourceJob: (      input: DeliveryAttemptInput & { readonly sourceJobId: string },    ) => Effect.Effect<DeliverySourceJobClaimResult, DeliveryAttemptRecordPersistenceError>;    readonly completeSourceJob: (      input: DeliveryAttemptCompletionInput,    ) => Effect.Effect<void, DeliveryAttemptRecordPersistenceError>;  }>()("t3code-relay/agentActivity/DeliveryAttempts") {} const SOURCE_JOB_CLAIM_LEASE_MINUTES = 10; function insertValues(  input: DeliveryAttemptInput,  id: string,  createdAt: string,): typeof relayDeliveryAttempts.$inferInsert {  return {    id,    createdAt,    userId: input.userId,    environmentId: input.environmentId,    threadId: input.threadId,    deviceId: input.deviceId,    kind: input.kind,    sourceJobId: input.sourceJobId ?? null,    tokenSuffix: input.token?.slice(-8) ?? null,    apnsStatus: input.apnsStatus ?? null,    apnsReason: input.apnsReason ?? null,    apnsId: input.apnsId ?? null,    transportError: input.transportError ?? null,  };} export const make = Effect.gen(function* () {  const db = yield* RelayDb.RelayDb;  const crypto = yield* Crypto.Crypto;   const isExpiredClaim = (claimedAt: string | null, now: DateTime.DateTime) => {    if (claimedAt === null) {      return true;    }    return Option.match(DateTime.make(claimedAt), {      onNone: () => true,      onSome: (dateTime) =>        now.epochMilliseconds - dateTime.epochMilliseconds >=        SOURCE_JOB_CLAIM_LEASE_MINUTES * 60 * 1_000,    });  };   return DeliveryAttempts.of({    record: Effect.fn("relay.delivery_attempts.record")(function* (input) {      yield* Effect.annotateCurrentSpan({        "relay.delivery.kind": input.kind,        ...(input.sourceJobId ? { "relay.delivery.job_id": input.sourceJobId } : {}),        ...(input.deviceId ? { "relay.mobile.device_id": input.deviceId } : {}),        ...(input.environmentId ? { "relay.environment_id": input.environmentId } : {}),        ...(input.threadId ? { "relay.thread_id": input.threadId } : {}),      });      yield* Effect.gen(function* () {        const id = yield* crypto.randomUUIDv4;        const createdAt = DateTime.formatIso(yield* DateTime.now);        yield* db.insert(relayDeliveryAttempts).values(insertValues(input, id, createdAt));      }).pipe(        Effect.mapError(          (cause) =>            new DeliveryAttemptRecordPersistenceError({              operation: "record",              sourceJobId: input.sourceJobId ?? null,              userId: input.userId,              environmentId: input.environmentId,              threadId: input.threadId,              deviceId: input.deviceId,              kind: input.kind,              cause,            }),        ),      );    }),    claimSourceJob: Effect.fn("relay.delivery_attempts.claim_source_job")(function* (input) {      yield* Effect.annotateCurrentSpan({        "relay.delivery.kind": input.kind,        "relay.delivery.job_id": input.sourceJobId,        ...(input.deviceId ? { "relay.mobile.device_id": input.deviceId } : {}),        ...(input.environmentId ? { "relay.environment_id": input.environmentId } : {}),        ...(input.threadId ? { "relay.thread_id": input.threadId } : {}),      });      return yield* Effect.gen(function* () {        const id = yield* crypto.randomUUIDv4;        const now = yield* DateTime.now;        const createdAt = DateTime.formatIso(now);        const inserted = yield* db          .insert(relayDeliveryAttempts)          .values(insertValues(input, id, createdAt))          .onConflictDoNothing({ target: relayDeliveryAttempts.sourceJobId })          .returning({ id: relayDeliveryAttempts.id });        if (inserted.length > 0) {          return "claimed";        }         const existing = yield* db          .select({            createdAt: relayDeliveryAttempts.createdAt,            apnsStatus: relayDeliveryAttempts.apnsStatus,            apnsReason: relayDeliveryAttempts.apnsReason,            apnsId: relayDeliveryAttempts.apnsId,            transportError: relayDeliveryAttempts.transportError,          })          .from(relayDeliveryAttempts)          .where(eq(relayDeliveryAttempts.sourceJobId, input.sourceJobId))          .limit(1);        const row = existing[0];        if (!row) {          return "in_flight";        }        if (          row.apnsStatus !== null ||          row.apnsReason !== null ||          row.apnsId !== null ||          row.transportError !== null        ) {          return "completed";        }        if (!isExpiredClaim(row.createdAt, now)) {          return "in_flight";        }         const reclaimed = yield* db          .update(relayDeliveryAttempts)          .set({            createdAt,          })          .where(            and(              eq(relayDeliveryAttempts.sourceJobId, input.sourceJobId),              eq(relayDeliveryAttempts.createdAt, row.createdAt),              isNull(relayDeliveryAttempts.apnsStatus),              isNull(relayDeliveryAttempts.apnsReason),              isNull(relayDeliveryAttempts.apnsId),              isNull(relayDeliveryAttempts.transportError),            ),          )          .returning({ id: relayDeliveryAttempts.id });        return reclaimed.length > 0 ? "claimed" : "in_flight";      }).pipe(        Effect.mapError(          (cause) =>            new DeliveryAttemptRecordPersistenceError({              operation: "claim-source-job",              sourceJobId: input.sourceJobId,              userId: input.userId,              environmentId: input.environmentId,              threadId: input.threadId,              deviceId: input.deviceId,              kind: input.kind,              cause,            }),        ),      );    }),    completeSourceJob: Effect.fn("relay.delivery_attempts.complete_source_job")(function* (input) {      yield* Effect.annotateCurrentSpan({ "relay.delivery.job_id": input.sourceJobId });      const completedAt = DateTime.formatIso(yield* DateTime.now);      yield* db        .update(relayDeliveryAttempts)        .set({          createdAt: completedAt,          apnsStatus: input.apnsStatus ?? null,          apnsReason: input.apnsReason ?? null,          apnsId: input.apnsId ?? null,          transportError: input.transportError ?? null,        })        .where(eq(relayDeliveryAttempts.sourceJobId, input.sourceJobId))        .pipe(          Effect.mapError(            (cause) =>              new DeliveryAttemptRecordPersistenceError({                operation: "complete-source-job",                sourceJobId: input.sourceJobId,                userId: null,                environmentId: null,                threadId: null,                deviceId: null,                kind: null,                cause,              }),          ),        );    }),  });}); export const layer = Layer.effect(DeliveryAttempts, make);