/**
 * Convex-backed Flue {@link PersistenceAdapter} for the agents package.
 *
 * Every store method maps to a public Convex function exported from
 * `fluePersistence.ts` under the `fluePersistence:<name>` path. The adapter
 * never imports generated backend bindings — it uses `makeFunctionReference` to
 * build typed references against explicit DTO shapes, so the file typechecks
 * against the installed runtime package without depending on codegen output.
 *
 * DTO conventions shared with the backend:
 *   - Submissions persist `inputJson` (the dispatch input or the direct input
 *     with image data stripped via `prepareDirectSubmission`) plus `chunksJson`
 *     for direct submissions (a `PersistedImageChunk[]`). Direct inputs are
 *     rehydrated with `hydratePersistedDirectSubmission`.
 *   - Conversation batches persist a numeric `offset` and `recordsJson`.
 *   - Event entries persist a numeric `offset` and `dataJson`.
 *   - Attachments carry `attachmentJson` and raw `bytes` as `ArrayBuffer`.
 *   - Run records persist `inputJson`/`resultJson`/`errorJson`/`traceCarrierJson`.
 *   - Append functions return `{ offset: number; appended: boolean }`; the
 *     client formats the offset into Flue's DS-compatible string and notifies
 *     in-process listeners only when `appended` is true (idempotent retries do
 *     not double-fire).
 *   - Read functions take a numeric `afterOffset` (parsed from the Flue offset
 *     string by the client); list functions take a decoded cursor.
 */

import { env } from "@code/env/server";
import {
  type AgentAttemptMarker,
  type AgentDispatchAdmission,
  type AgentDispatchReceipt,
  type AgentExecutionStore,
  type AgentSubmission,
  type AgentSubmissionStore,
  AttachmentConflictError,
  type AttachmentRef,
  type AttachmentStore,
  type ConversationProducerClaim,
  type ConversationRecord,
  type ConversationStreamBatch,
  type ConversationStreamIdentity,
  type ConversationStreamMeta,
  type ConversationStreamReadResult,
  type ConversationStreamStore,
  type CreateRunInput,
  DEFAULT_LIST_LIMIT,
  DEFAULT_READ_LIMIT,
  type DirectAgentSubmissionInput,
  type DispatchAgentSubmissionInput,
  type DispatchInput,
  type EndRunInput,
  type EventStreamMeta,
  type EventStreamReadResult,
  type EventStreamStore,
  type GetAttachmentInput,
  MAX_LIST_LIMIT,
  MAX_READ_LIMIT,
  type PersistedChunkRow,
  type PersistenceAdapter,
  type PersistenceStores,
  type PutAttachmentInput,
  type RunPointer,
  type RunRecord,
  type RunStore,
  type RunStatus,
  StreamListenerRegistry,
  type StoredAttachment,
  type SubmissionAttemptRef,
  type SubmissionClaimRef,
  type SubmissionDurability,
  type SubmissionSettlementObligation,
  type SubmissionSettledRecord,
  assertSupportedFlueSchemaVersion,
  clampLimit,
  copyAttachmentBytes,
  createSessionStorageKey,
  decodeRunCursor,
  encodeRunCursor,
  formatOffset,
  hydratePersistedDirectSubmission,
  parseOffset,
  prepareDirectSubmission,
  verifyAttachmentBytes,
} from "@flue/runtime/adapter";
import { ConvexHttpClient } from "convex/browser";
import {
  type FunctionArgs,
  type FunctionReference,
  type FunctionReturnType,
  makeFunctionReference,
} from "convex/server";

// ---------------------------------------------------------------------------
// Token + arg types
//
// All Convex arg types are declared as type aliases (not interfaces) so they
// satisfy Convex's `DefaultFunctionArgs = Record<string, unknown>` constraint.
// TypeScript does not give interfaces an implicit string index signature, but
// object-literal type aliases do.
// ---------------------------------------------------------------------------

/** Auth token present on every Convex function call. */
type TokenArgs = { readonly token: string; readonly [key: string]: unknown };

const TOKEN = (): string => env.FLUE_DB_TOKEN;

/** Trace carrier shape (W3C tracecontext). Not exported by Flue's adapter. */
type TraceCarrier = NonNullable<DirectAgentSubmissionInput["traceCarrier"]>;

// ---------------------------------------------------------------------------
// Shared DTO shapes (wire types returned by the Convex backend)
// ---------------------------------------------------------------------------

/** Common fields stored for every submission row. */
type SubmissionRow = {
  readonly sequence: number;
  readonly submissionId: string;
  readonly sessionKey: string;
  readonly kind: "dispatch" | "direct";
  readonly inputJson: string;
  readonly chunksJson: string | null;
  readonly status: "queued" | "running" | "terminalizing" | "settled";
  readonly acceptedAt: number;
  readonly canonicalReadyAt: number | null;
  readonly attemptId: string | null;
  readonly inputAppliedAt: number | null;
  readonly recoveryRequestedAt: number | null;
  readonly abortRequestedAt: number | null;
  readonly startedAt: number | null;
  readonly error: string | null;
  readonly attemptCount: number;
  readonly maxRetry: number;
  readonly timeoutAt: number;
  readonly ownerId: string | null;
  readonly leaseExpiresAt: number;
  readonly traceCarrierJson: string | null;
};

type SettlementObligationRow = {
  readonly submissionId: string;
  readonly sessionKey: string;
  readonly attemptId: string;
  readonly recordId: string;
  readonly recordJson: string;
};

type AttemptMarkerRow = {
  readonly submissionId: string;
  readonly attemptId: string;
  readonly createdAt: number;
};

type ConversationStreamRow = {
  readonly identity: ConversationStreamIdentity;
  readonly incarnation: string;
  readonly nextOffset: number;
  readonly producerId: string | null;
  readonly producerEpoch: number;
  readonly nextProducerSequence: number;
};

type ConversationBatchRow = {
  readonly offset: number;
  readonly recordsJson: string;
};

type EventEntryRow = {
  readonly offset: number;
  readonly dataJson: string;
};

type RunRow = {
  readonly runId: string;
  readonly workflowName: string;
  readonly status: RunStatus;
  readonly startedAt: string;
  readonly endedAt: string | null;
  readonly isError: boolean | null;
  readonly durationMs: number | null;
  readonly inputJson: string | null;
  readonly resultJson: string | null;
  readonly errorJson: string | null;
  readonly traceCarrierJson: string | null;
};

type RunPointerRow = {
  readonly runId: string;
  readonly workflowName: string;
  readonly status: RunStatus;
  readonly startedAt: string;
  readonly endedAt: string | null;
  readonly durationMs: number | null;
  readonly isError: boolean | null;
};

type AttachmentRefWire = {
  readonly id: string;
  readonly mimeType: string;
  readonly size: number;
  readonly digest: string;
  readonly filename?: string;
};

// ---------------------------------------------------------------------------
// Typed function references
//
// `makeFunctionReference` builds a typed reference to a public Convex function
// by name. We declare explicit arg/return DTOs so the client side typechecks
// without importing generated backend bindings.
// ---------------------------------------------------------------------------

const checkSchemaVersion = makeFunctionReference<"query", TokenArgs, string>(
  "fluePersistence:checkSchemaVersion",
);

const getSubmission = makeFunctionReference<
  "query",
  TokenArgs & { readonly submissionId: string },
  SubmissionRow | null
>("fluePersistence:getSubmission");

const hasUnsettledSubmissions = makeFunctionReference<
  "query",
  TokenArgs,
  boolean
>("fluePersistence:hasUnsettledSubmissions");

const listRunnableSubmissions = makeFunctionReference<
  "query",
  TokenArgs,
  SubmissionRow[]
>("fluePersistence:listRunnableSubmissions");

const listUnreadySubmissions = makeFunctionReference<
  "query",
  TokenArgs,
  SubmissionRow[]
>("fluePersistence:listUnreadySubmissions");

const listRunningSubmissions = makeFunctionReference<
  "query",
  TokenArgs,
  SubmissionRow[]
>("fluePersistence:listRunningSubmissions");

const listPendingSubmissionSettlements = makeFunctionReference<
  "query",
  TokenArgs,
  SettlementObligationRow[]
>("fluePersistence:listPendingSubmissionSettlements");

type ReplaceAttemptArgs = TokenArgs & {
  readonly submissionId: string;
  readonly attemptId: string;
  readonly nextAttemptId: string;
  readonly ownerId?: string;
  readonly leaseExpiresAt?: number;
};
const replaceSubmissionAttempt = makeFunctionReference<
  "mutation",
  ReplaceAttemptArgs,
  SubmissionRow | null
>("fluePersistence:replaceSubmissionAttempt");

/** Unified admission envelope — `kind` discriminates dispatch vs direct. */
type AdmitSubmissionEnvelope = {
  readonly kind: "dispatch" | "direct";
  readonly submissionId: string;
  readonly sessionKey: string;
  readonly acceptedAt: number;
  readonly inputJson: string;
  readonly chunksJson?: string;
  readonly traceCarrierJson?: string;
};
type AdmitSubmissionResponse = {
  readonly kind: "submission" | "retained_receipt" | "conflict";
  readonly submission?: SubmissionRow;
  readonly receipt?: AgentDispatchReceipt;
};
const admitSubmission = makeFunctionReference<
  "mutation",
  TokenArgs & { readonly input: AdmitSubmissionEnvelope },
  AdmitSubmissionResponse
>("fluePersistence:admitSubmission");

const markSubmissionCanonicalReady = makeFunctionReference<
  "mutation",
  TokenArgs & { readonly submissionId: string },
  SubmissionRow | null
>("fluePersistence:markSubmissionCanonicalReady");

type ClaimArgs = TokenArgs & SubmissionClaimRef;
const claimSubmission = makeFunctionReference<
  "mutation",
  ClaimArgs,
  SubmissionRow | null
>("fluePersistence:claimSubmission");

type MarkInputAppliedArgs = TokenArgs &
  SubmissionAttemptRef & {
    readonly maxRetry?: number;
    readonly timeoutAt?: number;
  };
const markSubmissionInputApplied = makeFunctionReference<
  "mutation",
  MarkInputAppliedArgs,
  boolean
>("fluePersistence:markSubmissionInputApplied");

type AttemptArgs = TokenArgs & SubmissionAttemptRef;
const requestSubmissionRecovery = makeFunctionReference<
  "mutation",
  AttemptArgs,
  boolean
>("fluePersistence:requestSubmissionRecovery");

const requestSessionAbort = makeFunctionReference<
  "mutation",
  TokenArgs & { readonly sessionKey: string },
  string[]
>("fluePersistence:requestSessionAbort");

const requeueSubmissionBeforeInputApplied = makeFunctionReference<
  "mutation",
  AttemptArgs,
  boolean
>("fluePersistence:requeueSubmissionBeforeInputApplied");

type ReserveSettlementArgs = TokenArgs &
  SubmissionAttemptRef & {
    readonly recordId: string;
    readonly recordJson: string;
  };
const reserveSubmissionSettlement = makeFunctionReference<
  "mutation",
  ReserveSettlementArgs,
  SettlementObligationRow | null
>("fluePersistence:reserveSubmissionSettlement");

const finalizeSubmissionSettlement = makeFunctionReference<
  "mutation",
  AttemptArgs & { readonly recordId: string },
  boolean
>("fluePersistence:finalizeSubmissionSettlement");

type SettleArgs = TokenArgs &
  SubmissionAttemptRef & {
    readonly outcome: "completed" | "failed";
    readonly errorJson?: string;
  };
const settleSubmission = makeFunctionReference<
  "mutation",
  SettleArgs,
  boolean
>("fluePersistence:settleSubmission");

const insertAttemptMarker = makeFunctionReference<
  "mutation",
  AttemptArgs,
  AttemptMarkerRow
>("fluePersistence:insertAttemptMarker");

const deleteAttemptMarker = makeFunctionReference<
  "mutation",
  AttemptArgs,
  void
>("fluePersistence:deleteAttemptMarker");

const listAttemptMarkers = makeFunctionReference<
  "query",
  TokenArgs,
  AttemptMarkerRow[]
>("fluePersistence:listAttemptMarkers");

const renewLeases = makeFunctionReference<
  "mutation",
  TokenArgs & { readonly ownerId: string; readonly submissionIds: string[] },
  void
>("fluePersistence:renewLeases");

const listExpiredSubmissions = makeFunctionReference<
  "query",
  TokenArgs,
  SubmissionRow[]
>("fluePersistence:listExpiredSubmissions");

// Conversation stream
type CreateStreamArgs = TokenArgs & {
  readonly path: string;
  readonly identity: ConversationStreamIdentity;
};
const createConversationStream = makeFunctionReference<
  "mutation",
  CreateStreamArgs,
  void
>("fluePersistence:createConversationStream");

const acquireConversationProducer = makeFunctionReference<
  "mutation",
  TokenArgs & { readonly path: string; readonly producerId: string },
  ConversationProducerClaim
>("fluePersistence:acquireConversationProducer");

type AppendBatchArgs = TokenArgs & {
  readonly path: string;
  readonly producerId: string;
  readonly producerEpoch: number;
  readonly incarnation: string;
  readonly producerSequence: number;
  readonly submission?: { readonly submissionId: string; readonly attemptId: string };
  readonly recordsJson: string;
};
type AppendResult = { readonly offset: number; readonly appended: boolean };
const appendConversationBatch = makeFunctionReference<
  "mutation",
  AppendBatchArgs,
  AppendResult
>("fluePersistence:appendConversationBatch");

type ReadBatchesArgs = TokenArgs & {
  readonly path: string;
  readonly afterOffset: number;
  readonly limit: number;
};
type ReadBatchesResult = {
  readonly batches: ConversationBatchRow[];
  readonly nextOffset: number;
  readonly upToDate: boolean;
};
const readConversationBatches = makeFunctionReference<
  "query",
  ReadBatchesArgs,
  ReadBatchesResult
>("fluePersistence:readConversationBatches");

const getConversationStreamMeta = makeFunctionReference<
  "query",
  TokenArgs & { readonly path: string },
  ConversationStreamRow | null
>("fluePersistence:getConversationStreamMeta");

const deleteConversationStream = makeFunctionReference<
  "mutation",
  TokenArgs & { readonly path: string },
  void
>("fluePersistence:deleteConversationStream");

// Event stream
const createEventStream = makeFunctionReference<
  "mutation",
  TokenArgs & { readonly path: string },
  void
>("fluePersistence:createEventStream");

type AppendEventArgs = TokenArgs & {
  readonly path: string;
  readonly dataJson: string;
};
const appendEventFn = makeFunctionReference<
  "mutation",
  AppendEventArgs,
  AppendResult
>("fluePersistence:appendEvent");

type AppendEventOnceArgs = TokenArgs & {
  readonly path: string;
  readonly key: string;
  readonly dataJson: string;
};
const appendEventOnce = makeFunctionReference<
  "mutation",
  AppendEventOnceArgs,
  AppendResult
>("fluePersistence:appendEventOnce");

type ReadEventsArgs = TokenArgs & {
  readonly path: string;
  readonly afterOffset: number;
  readonly limit: number;
};
type ReadEventsResult = {
  readonly events: EventEntryRow[];
  readonly nextOffset: number;
  readonly upToDate: boolean;
  readonly closed: boolean;
};
const readEventsFn = makeFunctionReference<
  "query",
  ReadEventsArgs,
  ReadEventsResult
>("fluePersistence:readEvents");

const closeEventStream = makeFunctionReference<
  "mutation",
  TokenArgs & { readonly path: string },
  void
>("fluePersistence:closeEventStream");

type EventStreamMetaRow = { readonly nextOffset: number; readonly closed: boolean };
const getEventStreamMeta = makeFunctionReference<
  "query",
  TokenArgs & { readonly path: string },
  EventStreamMetaRow | null
>("fluePersistence:getEventStreamMeta");

// Run store
type CreateRunArgs = TokenArgs & {
  readonly runId: string;
  readonly workflowName: string;
  readonly startedAt: string;
  readonly inputJson: string;
  readonly traceCarrierJson?: string;
};
const createRun = makeFunctionReference<"mutation", CreateRunArgs, void>(
  "fluePersistence:createRun",
);

type EndRunArgs = TokenArgs & {
  readonly runId: string;
  readonly endedAt: string;
  readonly isError: boolean;
  readonly durationMs: number;
  readonly resultJson?: string;
  readonly errorJson?: string;
};
const endRun = makeFunctionReference<"mutation", EndRunArgs, void>(
  "fluePersistence:endRun",
);

const getRun = makeFunctionReference<
  "query",
  TokenArgs & { readonly runId: string },
  RunRow | null
>("fluePersistence:getRun");

const lookupRun = makeFunctionReference<
  "query",
  TokenArgs & { readonly runId: string },
  { readonly runId: string; readonly workflowName: string } | null
>("fluePersistence:lookupRun");

type ListRunsArgs = TokenArgs & {
  readonly status?: RunStatus;
  readonly workflowName?: string;
  readonly limit: number;
  readonly cursor?: { readonly startedAt: string; readonly runId: string };
};
type ListRunsResult = {
  readonly rows: RunPointerRow[];
  readonly hasMore: boolean;
};
const listRuns = makeFunctionReference<"query", ListRunsArgs, ListRunsResult>(
  "fluePersistence:listRuns",
);

// Attachments
type PutAttachmentArgs = TokenArgs & {
  readonly streamPath: string;
  readonly conversationId: string;
  readonly attachment: AttachmentRefWire;
  readonly bytes: ArrayBuffer;
};
const putAttachment = makeFunctionReference<
  "mutation",
  PutAttachmentArgs,
  "inserted" | "existing" | "conflict"
>("fluePersistence:putAttachment");

type GetAttachmentArgs = TokenArgs & {
  readonly streamPath: string;
  readonly conversationId: string;
  readonly attachmentId: string;
};
type AttachmentWire = {
  readonly attachment: AttachmentRefWire;
  readonly bytes: ArrayBuffer;
};
const getAttachment = makeFunctionReference<
  "query",
  GetAttachmentArgs,
  AttachmentWire | null
>("fluePersistence:getAttachment");

const deleteAttachmentsForInstance = makeFunctionReference<
  "mutation",
  TokenArgs & { readonly streamPath: string },
  void
>("fluePersistence:deleteAttachmentsForInstance");

// ---------------------------------------------------------------------------
// Marshalling helpers
// ---------------------------------------------------------------------------

const safeJsonParse = <T>(text: string | null | undefined): T | undefined => {
  if (text === null || text === undefined || text === "") return undefined;
  return JSON.parse(text) as T;
};

const jsonStringify = (value: unknown): string => JSON.stringify(value);

const parseAcceptedAt = (value: string, label: string): number => {
  const ms = Date.parse(value);
  if (!Number.isFinite(ms)) {
    throw new Error(
      `[flue] ${label} produced non-finite acceptedAt: ${value}`,
    );
  }
  return ms;
};

/** Parse a Flue offset string into a numeric `afterOffset` for backend reads. */
const afterOffset = (offset: string | undefined): number =>
  offset === undefined ? -1 : parseOffset(offset);

const wireRefToAttachmentRef = (wire: AttachmentRefWire): AttachmentRef => ({
  id: wire.id,
  mimeType: wire.mimeType,
  size: wire.size,
  digest: wire.digest,
  ...(wire.filename === undefined ? {} : { filename: wire.filename }),
});

const attachmentRefToWire = (ref: AttachmentRef): AttachmentRefWire => ({
  id: ref.id,
  mimeType: ref.mimeType,
  size: ref.size,
  digest: ref.digest,
  ...(ref.filename === undefined ? {} : { filename: ref.filename }),
});

/**
 * Rehydrate a {@link SubmissionRow} into the {@link AgentSubmission} contract.
 * `dispatch` inputs round-trip through `inputJson`; `direct` inputs are split
 * into `inputJson` (images stripped) + `chunksJson` and rehydrated with Flue's
 * `hydratePersistedDirectSubmission`.
 */
const hydrateSubmission = (row: SubmissionRow): AgentSubmission => {
  const baseSubmission: Omit<AgentSubmission, "input"> = {
    sequence: row.sequence,
    submissionId: row.submissionId,
    sessionKey: row.sessionKey,
    kind: row.kind,
    status: row.status,
    acceptedAt: row.acceptedAt,
    canonicalReadyAt: row.canonicalReadyAt,
    attemptCount: row.attemptCount,
    maxRetry: row.maxRetry,
    timeoutAt: row.timeoutAt,
    leaseExpiresAt: row.leaseExpiresAt,
    ...(row.attemptId === null ? {} : { attemptId: row.attemptId }),
    ...(row.inputAppliedAt === null
      ? {}
      : { inputAppliedAt: row.inputAppliedAt }),
    ...(row.recoveryRequestedAt === null
      ? {}
      : { recoveryRequestedAt: row.recoveryRequestedAt }),
    ...(row.abortRequestedAt === null
      ? {}
      : { abortRequestedAt: row.abortRequestedAt }),
    ...(row.startedAt === null ? {} : { startedAt: row.startedAt }),
    ...(row.error === null ? {} : { error: row.error }),
    ...(row.ownerId === null ? {} : { ownerId: row.ownerId }),
  };

  const traceCarrier = safeJsonParse<TraceCarrier>(row.traceCarrierJson);

  if (row.kind === "dispatch") {
    const input = safeJsonParse<DispatchAgentSubmissionInput>(row.inputJson);
    if (input === undefined) {
      throw new Error(
        `[flue] persisted dispatch submission ${row.submissionId} has empty inputJson`,
      );
    }
    const inputWithCarrier =
      traceCarrier === undefined ? input : { ...input, traceCarrier };
    return { ...baseSubmission, input: inputWithCarrier };
  }

  const stripped = safeJsonParse<DirectAgentSubmissionInput>(row.inputJson);
  if (stripped === undefined) {
    throw new Error(
      `[flue] persisted direct submission ${row.submissionId} has empty inputJson`,
    );
  }
  const chunks = safeJsonParse<PersistedChunkRow[]>(row.chunksJson) ?? [];
  const hydrated = hydratePersistedDirectSubmission(stripped, chunks);
  const input =
    traceCarrier === undefined ? hydrated : { ...hydrated, traceCarrier };
  return { ...baseSubmission, input };
};

const hydrateObligation = (
  row: SettlementObligationRow,
): SubmissionSettlementObligation => {
  const record = safeJsonParse<SubmissionSettledRecord>(row.recordJson);
  if (record === undefined) {
    throw new Error(
      `[flue] persisted settlement obligation ${row.submissionId} has empty recordJson`,
    );
  }
  return {
    submissionId: row.submissionId,
    sessionKey: row.sessionKey,
    attemptId: row.attemptId,
    recordId: row.recordId,
    record,
  };
};

const toRunRecord = (row: RunRow): RunRecord => ({
  runId: row.runId,
  workflowName: row.workflowName,
  status: row.status,
  startedAt: row.startedAt,
  ...(row.endedAt === null ? {} : { endedAt: row.endedAt }),
  ...(row.isError === null ? {} : { isError: row.isError }),
  ...(row.durationMs === null ? {} : { durationMs: row.durationMs }),
  ...(row.inputJson === null ? {} : { input: safeJsonParse(row.inputJson) }),
  ...(row.resultJson === null ? {} : { result: safeJsonParse(row.resultJson) }),
  ...(row.errorJson === null ? {} : { error: safeJsonParse(row.errorJson) }),
  ...(row.traceCarrierJson === null
    ? {}
    : { traceCarrier: safeJsonParse(row.traceCarrierJson) }),
});

const toRunPointer = (row: RunPointerRow): RunPointer => ({
  runId: row.runId,
  workflowName: row.workflowName,
  status: row.status,
  startedAt: row.startedAt,
  ...(row.endedAt === null ? {} : { endedAt: row.endedAt }),
  ...(row.durationMs === null ? {} : { durationMs: row.durationMs }),
  ...(row.isError === null ? {} : { isError: row.isError }),
});

// ---------------------------------------------------------------------------
// Client wrapper
//
// A typed facade over ConvexHttpClient. The generic methods constrain the
// function reference to the proper FunctionReference kind ("query" / "mutation")
// and infer args + return type, so call sites stay type-safe without casts.
// ---------------------------------------------------------------------------

class ConvexClient {
  private readonly client: ConvexHttpClient;

  constructor(url: string) {
    this.client = new ConvexHttpClient(url);
  }

  query<F extends FunctionReference<"query">>(
    ref: F,
    args: FunctionArgs<F>,
  ): Promise<FunctionReturnType<F>> {
    return this.client.query(ref, args);
  }

  mutation<F extends FunctionReference<"mutation">>(
    ref: F,
    args: FunctionArgs<F>,
  ): Promise<FunctionReturnType<F>> {
    return this.client.mutation(ref, args);
  }
}

// ---------------------------------------------------------------------------
// AgentSubmissionStore
// ---------------------------------------------------------------------------

class ConvexAgentSubmissionStore implements AgentSubmissionStore {
  constructor(private readonly convex: ConvexClient) {}

  async getSubmission(submissionId: string): Promise<AgentSubmission | null> {
    const row = await this.convex.query(getSubmission, {
      token: TOKEN(),
      submissionId,
    });
    return row === null ? null : hydrateSubmission(row);
  }

  async hasUnsettledSubmissions(): Promise<boolean> {
    return this.convex.query(hasUnsettledSubmissions, { token: TOKEN() });
  }

  async listRunnableSubmissions(): Promise<AgentSubmission[]> {
    const rows = await this.convex.query(listRunnableSubmissions, {
      token: TOKEN(),
    });
    return rows.map(hydrateSubmission);
  }

  async listUnreadySubmissions(): Promise<AgentSubmission[]> {
    const rows = await this.convex.query(listUnreadySubmissions, {
      token: TOKEN(),
    });
    return rows.map(hydrateSubmission);
  }

  async listRunningSubmissions(): Promise<AgentSubmission[]> {
    const rows = await this.convex.query(listRunningSubmissions, {
      token: TOKEN(),
    });
    return rows.map(hydrateSubmission);
  }

  async listPendingSubmissionSettlements(): Promise<
    SubmissionSettlementObligation[]
  > {
    const rows = await this.convex.query(listPendingSubmissionSettlements, {
      token: TOKEN(),
    });
    return rows.map(hydrateObligation);
  }

  async replaceSubmissionAttempt(
    attempt: SubmissionAttemptRef,
    nextAttemptId: string,
    lease?: { ownerId: string; leaseExpiresAt: number },
  ): Promise<AgentSubmission | null> {
    const row = await this.convex.mutation(replaceSubmissionAttempt, {
      token: TOKEN(),
      submissionId: attempt.submissionId,
      attemptId: attempt.attemptId,
      nextAttemptId,
      ...(lease === undefined ? {} : lease),
    });
    return row === null ? null : hydrateSubmission(row);
  }

  async admitDispatch(input: DispatchInput): Promise<AgentDispatchAdmission> {
    const sessionKey = createSessionStorageKey(input.id, "default", "default");
    const submissionInput: Omit<
      DispatchAgentSubmissionInput,
      "kind" | "submissionId"
    > = {
      dispatchId: input.dispatchId,
      agent: input.agent,
      id: input.id,
      input: input.input,
      acceptedAt: input.acceptedAt,
    };
    const res = await this.convex.mutation(admitSubmission, {
      token: TOKEN(),
      input: {
        kind: "dispatch",
        submissionId: input.dispatchId,
        sessionKey,
        acceptedAt: parseAcceptedAt(input.acceptedAt, "admitDispatch"),
        inputJson: jsonStringify({
          kind: "dispatch",
          submissionId: input.dispatchId,
          ...submissionInput,
        }),
      },
    });
    if (res.kind === "submission" && res.submission !== undefined) {
      return {
        kind: "submission",
        submission: hydrateSubmission(res.submission),
      };
    }
    if (res.kind === "retained_receipt" && res.receipt !== undefined) {
      return { kind: "retained_receipt", receipt: res.receipt };
    }
    return { kind: "conflict" };
  }

  async admitDirect(
    input: DirectAgentSubmissionInput,
  ): Promise<AgentSubmission> {
    const sessionKey = createSessionStorageKey(input.id, "default", "default");
    const extracted = prepareDirectSubmission(input);
    const res = await this.convex.mutation(admitSubmission, {
      token: TOKEN(),
      input: {
        kind: "direct",
        submissionId: input.submissionId,
        sessionKey,
        acceptedAt: parseAcceptedAt(input.acceptedAt, "admitDirect"),
        inputJson: jsonStringify(extracted.value),
        chunksJson: jsonStringify(extracted.chunks),
        ...(input.traceCarrier === undefined
          ? {}
          : { traceCarrierJson: jsonStringify(input.traceCarrier) }),
      },
    });
    if (res.kind === "submission" && res.submission !== undefined) {
      return hydrateSubmission(res.submission);
    }
    // Idempotent direct admission should always surface the existing
    // submission; if the backend signals a non-submission result we cannot
    // satisfy the `AgentSubmission` return type.
    throw new Error(
      `[flue] admitDirect for ${input.submissionId} did not return a submission`,
    );
  }

  async markSubmissionCanonicalReady(
    submissionId: string,
  ): Promise<AgentSubmission | null> {
    const row = await this.convex.mutation(markSubmissionCanonicalReady, {
      token: TOKEN(),
      submissionId,
    });
    return row === null ? null : hydrateSubmission(row);
  }

  async claimSubmission(
    claim: SubmissionClaimRef,
  ): Promise<AgentSubmission | null> {
    const row = await this.convex.mutation(claimSubmission, {
      token: TOKEN(),
      submissionId: claim.submissionId,
      attemptId: claim.attemptId,
      ownerId: claim.ownerId,
      leaseExpiresAt: claim.leaseExpiresAt,
    });
    return row === null ? null : hydrateSubmission(row);
  }

  async markSubmissionInputApplied(
    attempt: SubmissionAttemptRef,
    durability?: SubmissionDurability,
  ): Promise<boolean> {
    return this.convex.mutation(markSubmissionInputApplied, {
      token: TOKEN(),
      submissionId: attempt.submissionId,
      attemptId: attempt.attemptId,
      ...(durability === undefined ? {} : durability),
    });
  }

  async requestSubmissionRecovery(
    attempt: SubmissionAttemptRef,
  ): Promise<boolean> {
    return this.convex.mutation(requestSubmissionRecovery, {
      token: TOKEN(),
      submissionId: attempt.submissionId,
      attemptId: attempt.attemptId,
    });
  }

  async requestSessionAbort(sessionKey: string): Promise<string[]> {
    return this.convex.mutation(requestSessionAbort, {
      token: TOKEN(),
      sessionKey,
    });
  }

  async requeueSubmissionBeforeInputApplied(
    attempt: SubmissionAttemptRef,
  ): Promise<boolean> {
    return this.convex.mutation(requeueSubmissionBeforeInputApplied, {
      token: TOKEN(),
      submissionId: attempt.submissionId,
      attemptId: attempt.attemptId,
    });
  }

  async reserveSubmissionSettlement(
    attempt: SubmissionAttemptRef,
    settlement: { recordId: string; record: SubmissionSettledRecord },
  ): Promise<SubmissionSettlementObligation | null> {
    const row = await this.convex.mutation(reserveSubmissionSettlement, {
      token: TOKEN(),
      submissionId: attempt.submissionId,
      attemptId: attempt.attemptId,
      recordId: settlement.recordId,
      recordJson: jsonStringify(settlement.record),
    });
    return row === null ? null : hydrateObligation(row);
  }

  async finalizeSubmissionSettlement(
    attempt: SubmissionAttemptRef,
    recordId: string,
  ): Promise<boolean> {
    return this.convex.mutation(finalizeSubmissionSettlement, {
      token: TOKEN(),
      submissionId: attempt.submissionId,
      attemptId: attempt.attemptId,
      recordId,
    });
  }

  async completeSubmission(attempt: SubmissionAttemptRef): Promise<boolean> {
    return this.convex.mutation(settleSubmission, {
      token: TOKEN(),
      submissionId: attempt.submissionId,
      attemptId: attempt.attemptId,
      outcome: "completed",
    });
  }

  async failSubmission(
    attempt: SubmissionAttemptRef,
    error: unknown,
  ): Promise<boolean> {
    return this.convex.mutation(settleSubmission, {
      token: TOKEN(),
      submissionId: attempt.submissionId,
      attemptId: attempt.attemptId,
      outcome: "failed",
      errorJson: jsonStringify(error),
    });
  }

  async insertAttemptMarker(attempt: SubmissionAttemptRef): Promise<void> {
    await this.convex.mutation(insertAttemptMarker, {
      token: TOKEN(),
      submissionId: attempt.submissionId,
      attemptId: attempt.attemptId,
    });
  }

  async deleteAttemptMarker(attempt: SubmissionAttemptRef): Promise<void> {
    await this.convex.mutation(deleteAttemptMarker, {
      token: TOKEN(),
      submissionId: attempt.submissionId,
      attemptId: attempt.attemptId,
    });
  }

  async listAttemptMarkers(): Promise<AgentAttemptMarker[]> {
    const rows = await this.convex.query(listAttemptMarkers, {
      token: TOKEN(),
    });
    return rows.map((row) => ({
      submissionId: row.submissionId,
      attemptId: row.attemptId,
      createdAt: row.createdAt,
    }));
  }

  async renewLeases(ownerId: string, submissionIds: string[]): Promise<void> {
    await this.convex.mutation(renewLeases, {
      token: TOKEN(),
      ownerId,
      submissionIds,
    });
  }

  async listExpiredSubmissions(): Promise<AgentSubmission[]> {
    const rows = await this.convex.query(listExpiredSubmissions, {
      token: TOKEN(),
    });
    return rows.map(hydrateSubmission);
  }
}

// ---------------------------------------------------------------------------
// ConversationStreamStore
// ---------------------------------------------------------------------------

class ConvexConversationStreamStore implements ConversationStreamStore {
  private readonly listeners = new StreamListenerRegistry();

  constructor(private readonly convex: ConvexClient) {}

  async createStream(
    path: string,
    identity: ConversationStreamIdentity,
  ): Promise<void> {
    await this.convex.mutation(createConversationStream, {
      token: TOKEN(),
      path,
      identity,
    });
  }

  async acquireProducer(
    path: string,
    producerId: string,
  ): Promise<ConversationProducerClaim> {
    return this.convex.mutation(acquireConversationProducer, {
      token: TOKEN(),
      path,
      producerId,
    });
  }

  async append(input: {
    path: string;
    producerId: string;
    producerEpoch: number;
    incarnation: string;
    producerSequence: number;
    submission?: { submissionId: string; attemptId: string };
    records: readonly ConversationRecord[];
  }): Promise<{ offset: string }> {
    const result = await this.convex.mutation(appendConversationBatch, {
      token: TOKEN(),
      path: input.path,
      producerId: input.producerId,
      producerEpoch: input.producerEpoch,
      incarnation: input.incarnation,
      producerSequence: input.producerSequence,
      ...(input.submission === undefined ? {} : { submission: input.submission }),
      recordsJson: jsonStringify(input.records),
    });
    if (result.appended) {
      this.listeners.notify(input.path);
    }
    return { offset: formatOffset(result.offset) };
  }

  async read(
    path: string,
    options?: { offset?: string; limit?: number },
  ): Promise<ConversationStreamReadResult> {
    const limit = clampLimit(
      options?.limit,
      DEFAULT_READ_LIMIT,
      MAX_READ_LIMIT,
    );
    const result = await this.convex.query(readConversationBatches, {
      token: TOKEN(),
      path,
      afterOffset: afterOffset(options?.offset),
      limit,
    });
    const batches: ConversationStreamBatch[] = result.batches.map((row) => ({
      offset: formatOffset(row.offset),
      records: safeJsonParse<ConversationRecord[]>(row.recordsJson) ?? [],
    }));
    return {
      batches,
      nextOffset: formatOffset(result.nextOffset),
      upToDate: result.upToDate,
    };
  }

  async getMeta(path: string): Promise<ConversationStreamMeta | null> {
    const row = await this.convex.query(getConversationStreamMeta, {
      token: TOKEN(),
      path,
    });
    if (row === null) return null;
    return {
      identity: row.identity,
      incarnation: row.incarnation,
      nextOffset: formatOffset(row.nextOffset),
      producerId: row.producerId,
      producerEpoch: row.producerEpoch,
      nextProducerSequence: row.nextProducerSequence,
    };
  }

  async delete(path: string): Promise<void> {
    await this.convex.mutation(deleteConversationStream, {
      token: TOKEN(),
      path,
    });
  }

  subscribe(path: string, listener: () => void): () => void {
    return this.listeners.subscribe(path, listener);
  }
}

// ---------------------------------------------------------------------------
// EventStreamStore
// ---------------------------------------------------------------------------

class ConvexEventStreamStore implements EventStreamStore {
  private readonly listeners = new StreamListenerRegistry();

  constructor(private readonly convex: ConvexClient) {}

  async createStream(path: string): Promise<void> {
    await this.convex.mutation(createEventStream, {
      token: TOKEN(),
      path,
    });
  }

  async appendEvent(path: string, event: unknown): Promise<string> {
    const result = await this.convex.mutation(appendEventFn, {
      token: TOKEN(),
      path,
      dataJson: jsonStringify(event),
    });
    if (result.appended) {
      this.listeners.notify(path);
    }
    return formatOffset(result.offset);
  }

  async appendEventOnce(
    path: string,
    key: string,
    event: unknown,
  ): Promise<string> {
    const result = await this.convex.mutation(appendEventOnce, {
      token: TOKEN(),
      path,
      key,
      dataJson: jsonStringify(event),
    });
    if (result.appended) {
      this.listeners.notify(path);
    }
    return formatOffset(result.offset);
  }

  async readEvents(
    path: string,
    opts?: { offset?: string; limit?: number },
  ): Promise<EventStreamReadResult> {
    const limit = clampLimit(opts?.limit, DEFAULT_READ_LIMIT, MAX_READ_LIMIT);
    const result = await this.convex.query(readEventsFn, {
      token: TOKEN(),
      path,
      afterOffset: afterOffset(opts?.offset),
      limit,
    });
    return {
      events: result.events.map((row) => ({
        offset: formatOffset(row.offset),
        data: safeJsonParse(row.dataJson),
      })),
      nextOffset: formatOffset(result.nextOffset),
      upToDate: result.upToDate,
      closed: result.closed,
    };
  }

  async closeStream(path: string): Promise<void> {
    await this.convex.mutation(closeEventStream, {
      token: TOKEN(),
      path,
    });
  }

  async getStreamMeta(path: string): Promise<EventStreamMeta | null> {
    const row = await this.convex.query(getEventStreamMeta, {
      token: TOKEN(),
      path,
    });
    if (row === null) return null;
    return {
      nextOffset: formatOffset(row.nextOffset),
      closed: row.closed,
    };
  }

  subscribe(path: string, listener: () => void): () => void {
    return this.listeners.subscribe(path, listener);
  }
}

// ---------------------------------------------------------------------------
// RunStore
// ---------------------------------------------------------------------------

class ConvexRunStore implements RunStore {
  constructor(private readonly convex: ConvexClient) {}

  async createRun(input: CreateRunInput): Promise<void> {
    await this.convex.mutation(createRun, {
      token: TOKEN(),
      runId: input.runId,
      workflowName: input.workflowName,
      startedAt: input.startedAt,
      inputJson: jsonStringify(input.input),
      ...(input.traceCarrier === undefined
        ? {}
        : { traceCarrierJson: jsonStringify(input.traceCarrier) }),
    });
  }

  async endRun(input: EndRunInput): Promise<void> {
    await this.convex.mutation(endRun, {
      token: TOKEN(),
      runId: input.runId,
      endedAt: input.endedAt,
      isError: input.isError,
      durationMs: input.durationMs,
      ...(input.result === undefined
        ? {}
        : { resultJson: jsonStringify(input.result) }),
      ...(input.error === undefined
        ? {}
        : { errorJson: jsonStringify(input.error) }),
    });
  }

  async getRun(runId: string): Promise<RunRecord | null> {
    const row = await this.convex.query(getRun, { token: TOKEN(), runId });
    return row === null ? null : toRunRecord(row);
  }

  async lookupRun(
    runId: string,
  ): Promise<{ runId: string; workflowName: string } | null> {
    return this.convex.query(lookupRun, { token: TOKEN(), runId });
  }

  async listRuns(opts?: {
    status?: RunStatus;
    workflowName?: string;
    limit?: number;
    cursor?: string;
  }): Promise<{ runs: RunPointer[]; nextCursor?: string }> {
    const limit = clampLimit(opts?.limit, DEFAULT_LIST_LIMIT, MAX_LIST_LIMIT);
    const decoded = decodeRunCursor(opts?.cursor);
    const result = await this.convex.query(listRuns, {
      token: TOKEN(),
      ...(opts?.status === undefined ? {} : { status: opts.status }),
      ...(opts?.workflowName === undefined
        ? {}
        : { workflowName: opts.workflowName }),
      limit,
      ...(decoded === undefined ? {} : { cursor: decoded }),
    });
    const runs = result.rows.map(toRunPointer);
    const last = runs[runs.length - 1];
    const nextCursor =
      result.hasMore && last !== undefined
        ? encodeRunCursor({ startedAt: last.startedAt, runId: last.runId })
        : undefined;
    return { runs, ...(nextCursor === undefined ? {} : { nextCursor }) };
  }
}

// ---------------------------------------------------------------------------
// AttachmentStore
// ---------------------------------------------------------------------------

class ConvexAttachmentStore implements AttachmentStore {
  constructor(private readonly convex: ConvexClient) {}

  async put(input: PutAttachmentInput): Promise<void> {
    // Defensive copy: callers hand us a Uint8Array view that may be backed by
    // a larger buffer; we encode an immutable ArrayBuffer slice so the wire
    // payload cannot be mutated after the call returns.
    const bytes = copyAttachmentBytes(input.bytes);
    const status = await this.convex.mutation(putAttachment, {
      token: TOKEN(),
      streamPath: input.streamPath,
      conversationId: input.conversationId,
      attachment: attachmentRefToWire(input.attachment),
      bytes: bytes.buffer.slice(
        bytes.byteOffset,
        bytes.byteOffset + bytes.byteLength,
      ) as ArrayBuffer,
    });
    if (status === "conflict") {
      throw new AttachmentConflictError({
        path: input.streamPath,
        attachmentId: input.attachment.id,
      });
    }
  }

  async get(input: GetAttachmentInput): Promise<StoredAttachment | null> {
    const row = await this.convex.query(getAttachment, {
      token: TOKEN(),
      streamPath: input.streamPath,
      conversationId: input.conversationId,
      attachmentId: input.attachmentId,
    });
    if (row === null) return null;
    const attachment = wireRefToAttachmentRef(row.attachment);
    const bytes = new Uint8Array(row.bytes);
    // Verify integrity before handing bytes back to the runtime; throws
    // AttachmentIntegrityError on size/digest mismatch.
    await verifyAttachmentBytes(attachment, bytes);
    return {
      attachment,
      bytes: copyAttachmentBytes(bytes),
    };
  }

  async deleteForInstance(streamPath: string): Promise<void> {
    await this.convex.mutation(deleteAttachmentsForInstance, {
      token: TOKEN(),
      streamPath,
    });
  }
}

// ---------------------------------------------------------------------------
// Adapter
// ---------------------------------------------------------------------------

class ConvexPersistenceAdapterImpl implements PersistenceAdapter {
  private readonly convex: ConvexClient;
  private migrated = false;

  constructor() {
    this.convex = new ConvexClient(env.CONVEX_URL);
  }

  async migrate(): Promise<void> {
    const version = await this.convex.query(checkSchemaVersion, {
      token: TOKEN(),
    });
    assertSupportedFlueSchemaVersion(version);
    this.migrated = true;
  }

  connect(): PersistenceStores {
    if (!this.migrated) {
      throw new Error(
        "[flue] ConvexPersistenceAdapter.connect() called before migrate() completed",
      );
    }
    const submissions = new ConvexAgentSubmissionStore(this.convex);
    const executionStore: AgentExecutionStore = { submissions };
    return {
      executionStore,
      runStore: new ConvexRunStore(this.convex),
      eventStreamStore: new ConvexEventStreamStore(this.convex),
      conversationStreamStore: new ConvexConversationStreamStore(this.convex),
      attachmentStore: new ConvexAttachmentStore(this.convex),
    };
  }

  close(): void {
    // ConvexHttpClient has no explicit teardown; the connection is HTTP-based
    // and resources are released on process exit.
  }
}

const adapter: PersistenceAdapter = new ConvexPersistenceAdapterImpl();

export default adapter;
