import {
  executionDatasetProfileIdSchema,
  executionDatasetResourceSchema,
  type ExecutionDatasetManifest,
  type ExecutionDatasetScope,
} from "@automate.ax/api-contract"
import { decode, type Encodable } from "@automate.ax/codec"
import { Shape, ShapeStream, type Row, type Value } from "@electric-sql/client"
import { PGlite, type Results } from "@electric-sql/pglite"
import { pg_trgm } from "@electric-sql/pglite/contrib/pg_trgm"
import { parse } from "pgsql-parser"
import { lock } from "proper-lockfile"
import { createHash, randomUUID } from "node:crypto"
import fs from "node:fs/promises"
import os from "node:os"
import path from "node:path"
import { setTimeout } from "node:timers/promises"
import { z } from "zod"
import { sessionApiClient } from "./api-client"
import { env } from "./env"
import {
  loadExecutionDataProfileSync,
  loadSavedTokenSync,
  saveExecutionDataProfileSync,
} from "./session"

const EXECUTION_DATA_ROOT = path.join(os.homedir(), ".automate", "data")
const INLINE_BINARY_LIMIT = 64 * 1024
const LOCAL_EXECUTION_DATASET_VERSION = 11
const EXECUTION_DATA_SEARCH_SCHEMA_VERSION = 3
const EXECUTION_DATA_CORRUPTION_CODES = new Set(["XX001", "XX002"])
const EXECUTION_DATA_LOCK_STALE_MS = 120_000
const EXECUTION_DATA_LOCK_UPDATE_MS = 30_000
const EXECUTION_DATA_LOCK_RETRY_MS = 50
const EXECUTION_DATA_RECOVERY = {
  localDatabaseRebuilt: true,
  productionDataAffected: false,
} as const

const LOCAL_EXECUTION_DATASET_RESOURCE_SCHEMA = executionDatasetResourceSchema
type LocalExecutionDatasetResource = z.infer<
  typeof LOCAL_EXECUTION_DATASET_RESOURCE_SCHEMA
>
const EXECUTION_DATA_RESOURCE_STATUS_SCHEMA = z.record(
  LOCAL_EXECUTION_DATASET_RESOURCE_SCHEMA,
  z.object({
    offset: z.string(),
    rows: z.number().int().nonnegative(),
    syncedAt: z.iso.datetime({ offset: true }),
  }),
)

const EXECUTION_DATA_BOOTSTRAP_SQL = `
CREATE SCHEMA IF NOT EXISTS automate_internal;
CREATE TABLE IF NOT EXISTS automate_internal.sync_state (
  singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton),
  profile_id text NOT NULL,
  schema_version integer NOT NULL,
  organization_id text,
  project_id text,
  since_at timestamptz,
  synced_at timestamptz NOT NULL
);
CREATE TABLE IF NOT EXISTS automate_internal.resource_sync_state (
  resource text PRIMARY KEY,
  shape_offset text NOT NULL,
  synced_at timestamptz NOT NULL
);
CREATE TABLE IF NOT EXISTS automate_internal.local_state (
  singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton),
  search_schema_version integer NOT NULL
);
`

export const DEFAULT_EXECUTION_DATA_SINCE = "7d"

type ElectricRow = Row<Uint8Array>

interface ResourceDefinition {
  columns: readonly string[]
  jsonColumns?: readonly string[]
  table: string
}

const RESOURCE_DEFINITIONS = {
  "action-attempts": {
    columns: [
      "id",
      "action_invocation_id",
      "delivery_id",
      "attempt",
      "status",
      "error",
      "retry_at",
      "started_at",
      "completed_at",
    ],
    jsonColumns: ["error"],
    table: "action_attempts",
  },
  actions: {
    columns: [
      "id",
      "max_attempts",
      "context_id",
      "name",
      "description",
      "slot",
      "scope_path",
      "status",
      "created_at",
      "derivation",
      "completed_at",
      "error",
      "skip_reason",
      "action_dependency_ids",
      "event_dependency_ids",
      "signal_dependency_ids",
      "outcome_seq",
      "output_encoded",
      "output",
      "output_sensitivity",
      "replay_safety",
    ],
    jsonColumns: ["derivation", "error", "output", "output_sensitivity"],
    table: "actions",
  },
  automations: {
    columns: [
      "id",
      "project_id",
      "identity_key",
      "enabled",
      "created_at",
      "updated_at",
    ],
    table: "automations",
  },
  "automation-invocations": {
    columns: [
      "id",
      "automation_id",
      "entrypoint",
      "idempotency_key",
      "run_at",
      "event_id",
      "processed_at",
      "created_at",
    ],
    table: "automation_invocations",
  },
  contexts: {
    columns: [
      "id",
      "automation_id",
      "status",
      "created_at",
      "significant",
      "settled_at",
      "error",
    ],
    jsonColumns: ["error"],
    table: "contexts",
  },
  "context-parents": {
    columns: [
      "context_id",
      "parent_context_id",
      "position",
      "is_primary",
      "inherits_frontier",
    ],
    table: "context_parents",
  },
  "event-links": {
    columns: [
      "id",
      "context_id",
      "event_id",
      "event_subscription_id",
      "hook_slot",
      "scope_path",
      "outcome_seq",
      "config",
    ],
    jsonColumns: ["config"],
    table: "event_links",
  },
  events: {
    columns: [
      "id",
      "event_source_id",
      "event_type",
      "ingest_seq",
      "created_at",
      "payload_encoded",
      "payload",
    ],
    jsonColumns: ["payload"],
    table: "events",
  },
  logs: {
    columns: [
      "id",
      "context_id",
      "action_invocation_id",
      "attempt",
      "log_index",
      "level",
      "message",
      "created_at",
      "fields_encoded",
      "fields",
      "sensitivity",
    ],
    jsonColumns: ["fields", "sensitivity"],
    table: "logs",
  },
  outputs: {
    columns: [
      "id",
      "context_id",
      "action_invocation_id",
      "output_index",
      "output_seq",
      "created_at",
      "type",
      "data",
      "value_encoded",
      "value",
      "value_sensitivity",
    ],
    jsonColumns: ["data", "value", "value_sensitivity"],
    table: "outputs",
  },
  "signal-coordinators": {
    columns: [
      "id",
      "automation_id",
      "configuration_hash",
      "primary_position",
      "scope_path",
      "slot",
    ],
    table: "signal_coordinators",
  },
  "signal-decisions": {
    columns: [
      "id",
      "context_id",
      "coordinator_id",
      "partition_id",
      "created_at",
    ],
    table: "signal_decisions",
  },
  "signal-decision-offers": {
    columns: ["decision_id", "offer_id", "position"],
    table: "signal_decision_offers",
  },
  "signal-invocations": {
    columns: [
      "id",
      "context_id",
      "slot",
      "scope_path",
      "status",
      "created_at",
      "derivation",
      "error",
      "selected_index",
      "action_dependency_ids",
      "event_dependency_ids",
      "signal_dependency_ids",
      "outcome_seq",
      "output_encoded",
      "output",
    ],
    jsonColumns: ["derivation", "error", "output"],
    table: "signal_invocations",
  },
  "signal-offers": {
    columns: [
      "id",
      "context_id",
      "coordinator_id",
      "decision_id",
      "lane",
      "status",
      "selected",
      "created_at",
      "derivation",
      "ready_at",
      "expires_at",
      "burst_ends_at",
      "outcome_seq",
      "action_dependency_ids",
      "event_dependency_ids",
      "signal_dependency_ids",
      "partition_id",
      "settled_at",
      "key_hash",
      "key_encoded",
      "key",
    ],
    jsonColumns: ["derivation", "key"],
    table: "signal_offers",
  },
  "signal-partitions": {
    columns: [
      "id",
      "coordinator_id",
      "created_at",
      "key_hash",
      "key_encoded",
      "key",
      "state",
      "state_expires_at",
      "wake_at",
    ],
    jsonColumns: ["key", "state"],
    table: "signal_partitions",
  },
  projects: {
    columns: [
      "id",
      "organization_id",
      "name",
      "active_deployment_id",
      "created_at",
    ],
    table: "projects",
  },
} as const satisfies Record<LocalExecutionDatasetResource, ResourceDefinition>

const RESOURCE_PRIMARY_KEYS = {
  "action-attempts": ["id"],
  actions: ["id"],
  automations: ["id"],
  "automation-invocations": ["id"],
  contexts: ["id"],
  "context-parents": ["context_id", "parent_context_id"],
  "event-links": ["id"],
  events: ["id"],
  logs: ["id"],
  outputs: ["id"],
  projects: ["id"],
  "signal-coordinators": ["id"],
  "signal-decisions": ["id"],
  "signal-decision-offers": ["decision_id", "offer_id"],
  "signal-invocations": ["id"],
  "signal-offers": ["id"],
  "signal-partitions": ["id"],
} as const satisfies Record<
  LocalExecutionDatasetResource,
  readonly (typeof RESOURCE_DEFINITIONS)[LocalExecutionDatasetResource]["columns"][number][]
>

export const EXECUTION_DATA_SCHEMA_SQL = `
CREATE SCHEMA IF NOT EXISTS automate;
CREATE SCHEMA IF NOT EXISTS automate_internal;
CREATE SCHEMA IF NOT EXISTS automate_extensions;
CREATE EXTENSION IF NOT EXISTS pg_trgm WITH SCHEMA automate_extensions;

CREATE TABLE IF NOT EXISTS automate.projects (
  id text PRIMARY KEY,
  organization_id text NOT NULL,
  name text NOT NULL,
  active_deployment_id text,
  created_at timestamptz NOT NULL
);
CREATE TABLE IF NOT EXISTS automate.automations (
  id text PRIMARY KEY,
  project_id text NOT NULL,
  identity_key text NOT NULL,
  enabled boolean NOT NULL,
  created_at timestamptz NOT NULL,
  updated_at timestamptz NOT NULL
);
CREATE TABLE IF NOT EXISTS automate.automation_invocations (
  id text PRIMARY KEY,
  automation_id text NOT NULL,
  entrypoint text NOT NULL,
  idempotency_key text NOT NULL,
  run_at timestamptz NOT NULL,
  event_id text,
  processed_at timestamptz,
  created_at timestamptz NOT NULL
);
CREATE TABLE IF NOT EXISTS automate.contexts (
  id text PRIMARY KEY,
  automation_id text NOT NULL,
  status text NOT NULL,
  created_at timestamptz NOT NULL,
  significant boolean NOT NULL DEFAULT false,
  settled_at timestamptz,
  error jsonb
);
CREATE TABLE IF NOT EXISTS automate.context_parents (
  context_id text NOT NULL,
  parent_context_id text NOT NULL,
  position integer NOT NULL,
  is_primary boolean NOT NULL,
  inherits_frontier boolean NOT NULL,
  PRIMARY KEY (context_id, parent_context_id)
);
CREATE TABLE IF NOT EXISTS automate.event_links (
  id text PRIMARY KEY,
  context_id text NOT NULL,
  event_id text NOT NULL,
  event_subscription_id text,
  hook_slot integer NOT NULL,
  scope_path integer[] NOT NULL,
  outcome_seq bigint NOT NULL,
  config jsonb NOT NULL
);
CREATE TABLE IF NOT EXISTS automate.events (
  id text PRIMARY KEY,
  event_source_id text NOT NULL,
  event_type text NOT NULL,
  ingest_seq bigint NOT NULL,
  created_at timestamptz NOT NULL,
  payload_encoded bytea NOT NULL,
  payload jsonb NOT NULL
);
CREATE TABLE IF NOT EXISTS automate.actions (
  id text PRIMARY KEY,
  context_id text NOT NULL,
  name text NOT NULL,
  description text,
  slot integer NOT NULL,
  scope_path integer[] NOT NULL,
  status text NOT NULL,
  created_at timestamptz NOT NULL,
  completed_at timestamptz,
  derivation jsonb,
  error jsonb,
  skip_reason text,
  action_dependency_ids text[] NOT NULL,
  event_dependency_ids text[] NOT NULL,
  signal_dependency_ids text[] NOT NULL,
  outcome_seq bigint,
  output_encoded bytea,
  output jsonb,
  output_sensitivity jsonb,
  max_attempts integer NOT NULL DEFAULT 3,
  replay_safety text
);
CREATE TABLE IF NOT EXISTS automate.action_attempts (
  id text PRIMARY KEY,
  action_invocation_id text NOT NULL,
  delivery_id text NOT NULL,
  attempt integer NOT NULL,
  status text NOT NULL,
  error jsonb,
  retry_at timestamptz,
  started_at timestamptz NOT NULL,
  completed_at timestamptz
);
CREATE TABLE IF NOT EXISTS automate.outputs (
  id text PRIMARY KEY,
  context_id text NOT NULL,
  action_invocation_id text NOT NULL,
  output_index integer NOT NULL,
  output_seq bigint NOT NULL,
  created_at timestamptz NOT NULL,
  type text,
  data jsonb,
  value_encoded bytea NOT NULL,
  value jsonb NOT NULL,
  value_sensitivity jsonb
);
CREATE TABLE IF NOT EXISTS automate.signal_coordinators (
  id text PRIMARY KEY,
  automation_id text NOT NULL,
  configuration_hash text NOT NULL,
  primary_position text,
  scope_path integer[] NOT NULL,
  slot integer NOT NULL
);
CREATE TABLE IF NOT EXISTS automate.signal_decisions (
  id text PRIMARY KEY,
  context_id text NOT NULL,
  coordinator_id text NOT NULL,
  partition_id text,
  created_at timestamptz NOT NULL
);
CREATE TABLE IF NOT EXISTS automate.signal_decision_offers (
  decision_id text NOT NULL,
  offer_id text NOT NULL,
  position integer NOT NULL,
  PRIMARY KEY (decision_id, offer_id)
);
CREATE TABLE IF NOT EXISTS automate.signal_invocations (
  id text PRIMARY KEY,
  context_id text NOT NULL,
  slot integer NOT NULL,
  scope_path integer[] NOT NULL,
  status text NOT NULL,
  created_at timestamptz NOT NULL,
  derivation jsonb,
  error jsonb,
  selected_index integer,
  action_dependency_ids text[] NOT NULL,
  event_dependency_ids text[] NOT NULL,
  signal_dependency_ids text[] NOT NULL,
  outcome_seq bigint,
  output_encoded bytea,
  output jsonb
);
CREATE TABLE IF NOT EXISTS automate.signal_offers (
  id text PRIMARY KEY,
  context_id text NOT NULL,
  coordinator_id text NOT NULL,
  decision_id text,
  lane text NOT NULL,
  status text NOT NULL,
  selected boolean NOT NULL,
  created_at timestamptz NOT NULL,
  derivation jsonb,
  ready_at timestamptz,
  expires_at timestamptz,
  burst_ends_at timestamptz,
  outcome_seq bigint NOT NULL,
  action_dependency_ids text[] NOT NULL,
  event_dependency_ids text[] NOT NULL,
  signal_dependency_ids text[] NOT NULL,
  partition_id text,
  settled_at timestamptz,
  key_hash text NOT NULL,
  key_encoded bytea NOT NULL,
  key jsonb NOT NULL
);
CREATE TABLE IF NOT EXISTS automate.signal_partitions (
  id text PRIMARY KEY,
  coordinator_id text NOT NULL,
  created_at timestamptz NOT NULL,
  key_hash text NOT NULL,
  key_encoded bytea NOT NULL,
  key jsonb NOT NULL,
  state jsonb,
  state_expires_at timestamptz,
  wake_at timestamptz
);
CREATE TABLE IF NOT EXISTS automate.logs (
  id text PRIMARY KEY,
  context_id text NOT NULL,
  action_invocation_id text NOT NULL,
  attempt integer NOT NULL,
  log_index integer NOT NULL,
  level text NOT NULL,
  message text NOT NULL,
  created_at timestamptz NOT NULL,
  fields_encoded bytea,
  fields jsonb,
  sensitivity jsonb
);
CREATE TABLE IF NOT EXISTS automate.search_documents (
  document_id text PRIMARY KEY,
  resource text NOT NULL,
  resource_id text NOT NULL,
  context_id text,
  project_id text,
  automation_id text,
  provider text,
  title text NOT NULL,
  content text NOT NULL,
  metadata text NOT NULL,
  identifiers text NOT NULL,
  created_at timestamptz,
  search_vector tsvector GENERATED ALWAYS AS (
    setweight(to_tsvector('simple', coalesce(title, '')), 'A') ||
    setweight(to_tsvector('simple', coalesce(content, '')), 'B') ||
    setweight(to_tsvector('simple', coalesce(metadata, '')), 'C') ||
    setweight(to_tsvector('simple', coalesce(identifiers, '')), 'D')
  ) STORED
);

CREATE INDEX IF NOT EXISTS automations_project_id_idx ON automate.automations (project_id);
CREATE INDEX IF NOT EXISTS automation_invocations_action_idx ON automate.automation_invocations (idempotency_key, run_at);
CREATE INDEX IF NOT EXISTS automation_invocations_event_idx ON automate.automation_invocations (event_id);
CREATE INDEX IF NOT EXISTS contexts_automation_created_idx ON automate.contexts (automation_id, created_at DESC);
CREATE INDEX IF NOT EXISTS context_parents_context_idx ON automate.context_parents (context_id);
CREATE INDEX IF NOT EXISTS context_parents_parent_idx ON automate.context_parents (parent_context_id);
CREATE INDEX IF NOT EXISTS event_links_context_idx ON automate.event_links (context_id);
CREATE INDEX IF NOT EXISTS actions_context_created_idx ON automate.actions (context_id, created_at);
CREATE INDEX IF NOT EXISTS outputs_context_seq_idx ON automate.outputs (context_id, output_seq);
CREATE INDEX IF NOT EXISTS signal_coordinators_automation_idx ON automate.signal_coordinators (automation_id, scope_path, slot);
CREATE INDEX IF NOT EXISTS signal_decisions_context_idx ON automate.signal_decisions (context_id, created_at);
CREATE INDEX IF NOT EXISTS signal_decision_offers_decision_idx ON automate.signal_decision_offers (decision_id, position);
CREATE INDEX IF NOT EXISTS signal_invocations_context_idx ON automate.signal_invocations (context_id, created_at);
CREATE INDEX IF NOT EXISTS signal_offers_context_idx ON automate.signal_offers (context_id, created_at);
CREATE INDEX IF NOT EXISTS signal_offers_coordinator_idx ON automate.signal_offers (coordinator_id, decision_id);
CREATE INDEX IF NOT EXISTS signal_partitions_coordinator_idx ON automate.signal_partitions (coordinator_id, key_hash);
CREATE INDEX IF NOT EXISTS events_type_created_idx ON automate.events (event_type, created_at DESC);
CREATE INDEX IF NOT EXISTS logs_context_created_idx ON automate.logs (context_id, created_at);
CREATE INDEX IF NOT EXISTS logs_level_created_idx ON automate.logs (level, created_at DESC);
CREATE INDEX IF NOT EXISTS search_documents_vector_idx ON automate.search_documents USING gin (search_vector);
CREATE INDEX IF NOT EXISTS search_documents_title_trgm_idx ON automate.search_documents USING gin (title automate_extensions.gin_trgm_ops);
CREATE INDEX IF NOT EXISTS search_documents_content_trgm_idx ON automate.search_documents USING gin (content automate_extensions.gin_trgm_ops);
CREATE INDEX IF NOT EXISTS search_documents_identifiers_trgm_idx ON automate.search_documents USING gin (identifiers automate_extensions.gin_trgm_ops);
CREATE INDEX IF NOT EXISTS search_documents_context_created_idx ON automate.search_documents (context_id, created_at DESC);

CREATE OR REPLACE VIEW automate.context_ancestors AS
WITH RECURSIVE ancestors(context_id, ancestor_context_id, depth, path) AS (
  SELECT id, id, 0, ARRAY[id] FROM automate.contexts
  UNION ALL
  SELECT ancestors.context_id, parents.parent_context_id, ancestors.depth + 1,
    ancestors.path || parents.parent_context_id
  FROM ancestors
  INNER JOIN automate.context_parents AS parents
    ON parents.context_id = ancestors.ancestor_context_id
  WHERE NOT parents.parent_context_id = ANY(ancestors.path)
)
SELECT context_id, ancestor_context_id, depth FROM ancestors;

CREATE OR REPLACE VIEW automate.context_descendants AS
WITH RECURSIVE descendants(context_id, descendant_context_id, depth, path) AS (
  SELECT id, id, 0, ARRAY[id] FROM automate.contexts
  UNION ALL
  SELECT descendants.context_id, parents.context_id, descendants.depth + 1,
    descendants.path || parents.context_id
  FROM descendants
  INNER JOIN automate.context_parents AS parents
    ON parents.parent_context_id = descendants.descendant_context_id
  WHERE NOT parents.context_id = ANY(descendants.path)
)
SELECT context_id, descendant_context_id, depth FROM descendants;

CREATE OR REPLACE VIEW automate.root_contexts AS
SELECT
  contexts.*,
  automations.project_id,
  automations.identity_key AS automation_identity_key,
  projects.organization_id,
  projects.name AS project_name
FROM automate.contexts
INNER JOIN automate.automations ON automations.id = contexts.automation_id
INNER JOIN automate.projects ON projects.id = automations.project_id
WHERE NOT EXISTS (
  SELECT 1 FROM automate.context_parents
  WHERE context_parents.context_id = contexts.id
);

CREATE OR REPLACE VIEW automate.signal_outcomes AS
SELECT
  signals.*,
  decisions.id AS decision_id,
  coalesce(partitions.coordinator_id, decisions.coordinator_id) AS coordinator_id,
  coordinators.configuration_hash AS coordinator_configuration_hash,
  coordinators.primary_position,
  coalesce(selected_offers.selected_offer_ids, ARRAY[]::text[]) AS selected_offer_ids,
  coalesce(selected_offers.selected_lanes, ARRAY[]::text[]) AS selected_lanes
FROM automate.signal_invocations AS signals
LEFT JOIN automate.contexts AS contexts ON contexts.id = signals.context_id
LEFT JOIN automate.signal_decisions AS decisions
  ON decisions.context_id = signals.context_id
LEFT JOIN automate.signal_partitions AS partitions
  ON partitions.id = decisions.partition_id
LEFT JOIN automate.signal_coordinators AS coordinators
  ON coordinators.id = coalesce(partitions.coordinator_id, decisions.coordinator_id)
LEFT JOIN LATERAL (
  SELECT
    array_agg(selected.id ORDER BY selected.position, selected.created_at, selected.id) AS selected_offer_ids,
    array_agg(selected.lane ORDER BY selected.position, selected.created_at, selected.id) AS selected_lanes
  FROM (
    SELECT offers.id, offers.lane, offers.created_at, membership.position
    FROM automate.signal_decision_offers AS membership
    INNER JOIN automate.signal_offers AS offers
      ON offers.id = membership.offer_id
    WHERE membership.decision_id = decisions.id

    UNION ALL

    -- DEPRECATED: supports datasets captured from a pre-AAX-332 process during
    -- rolling deployment. Remove after one full production release.
    SELECT
      offers.id,
      offers.lane,
      offers.created_at,
      row_number() over (ORDER BY offers.lane, offers.created_at, offers.id) - 1
    FROM automate.signal_offers AS offers
    WHERE offers.decision_id = decisions.id
      AND offers.selected
      AND NOT EXISTS (
        SELECT 1
        FROM automate.signal_decision_offers AS membership
        WHERE membership.decision_id = decisions.id
      )
  ) AS selected
) AS selected_offers ON true;

CREATE TABLE IF NOT EXISTS automate_internal.sync_state (
  singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton),
  profile_id text NOT NULL,
  schema_version integer NOT NULL,
  organization_id text,
  project_id text,
  since_at timestamptz,
  synced_at timestamptz NOT NULL
);
CREATE TABLE IF NOT EXISTS automate_internal.resource_sync_state (
  resource text PRIMARY KEY,
  shape_offset text NOT NULL,
  synced_at timestamptz NOT NULL
);
CREATE TABLE IF NOT EXISTS automate_internal.local_state (
  singleton boolean PRIMARY KEY DEFAULT true CHECK (singleton),
  search_schema_version integer NOT NULL
);

DO $$ BEGIN
  CREATE ROLE automate_query;
EXCEPTION WHEN duplicate_object THEN NULL;
END $$;
REVOKE ALL ON SCHEMA automate, automate_internal, automate_extensions FROM PUBLIC;
REVOKE ALL ON SCHEMA public FROM PUBLIC, automate_query;
REVOKE ALL ON ALL TABLES IN SCHEMA automate, automate_internal FROM PUBLIC;
REVOKE ALL ON ALL FUNCTIONS IN SCHEMA automate, automate_extensions, public FROM PUBLIC, automate_query;
GRANT USAGE ON SCHEMA automate, automate_extensions TO automate_query;
GRANT SELECT ON ALL TABLES IN SCHEMA automate TO automate_query;
GRANT EXECUTE ON ALL FUNCTIONS IN SCHEMA automate_extensions TO automate_query;
`

export const EXECUTION_DATA_SEARCH_INSERT_SQL = `
WITH context_metadata AS (
  SELECT
    contexts.id AS context_id,
    automations.id AS automation_id,
    automations.identity_key AS automation_identity_key,
    projects.id AS project_id,
    projects.name AS project_name,
    providers.provider
  FROM automate.contexts
  INNER JOIN automate.automations ON automations.id = contexts.automation_id
  INNER JOIN automate.projects ON projects.id = automations.project_id
  LEFT JOIN LATERAL (
    SELECT string_agg(
      DISTINCT split_part(events.event_type, '.', 1),
      ' ' ORDER BY split_part(events.event_type, '.', 1)
    ) AS provider
    FROM automate.event_links
    INNER JOIN automate.events ON events.id = event_links.event_id
    WHERE event_links.context_id = contexts.id
  ) AS providers ON true
), documents (
  document_id,
  resource,
  resource_id,
  context_id,
  project_id,
  automation_id,
  provider,
  title,
  content,
  metadata,
  identifiers,
  created_at
) AS (
SELECT
  'projects:' || projects.id,
  'projects',
  projects.id,
  NULL,
  projects.id,
  NULL,
  NULL,
  projects.name,
  projects.name,
  concat_ws(' ', projects.organization_id, projects.name),
  concat_ws(' ', projects.id, projects.organization_id),
  projects.created_at
FROM automate.projects
UNION ALL
SELECT
  'automations:' || automations.id,
  'automations',
  automations.id,
  NULL,
  projects.id,
  automations.id,
  NULL,
  automations.identity_key,
  concat_ws(' ', automations.identity_key, CASE WHEN automations.enabled THEN 'enabled' ELSE 'disabled' END),
  concat_ws(' ', projects.name, automations.identity_key),
  concat_ws(' ', automations.id, projects.id, projects.organization_id),
  automations.updated_at
FROM automate.automations
INNER JOIN automate.projects ON projects.id = automations.project_id
UNION ALL
SELECT
  'automation-invocations:' || invocations.id,
  'automation-invocations',
  invocations.id,
  actions.context_id,
  metadata.project_id,
  invocations.automation_id,
  metadata.provider,
  concat_ws(' ', 'scheduled invocation', invocations.entrypoint),
  concat_ws(E'\n', invocations.entrypoint, invocations.run_at::text, invocations.processed_at::text),
  concat_ws(' ', metadata.project_name, metadata.automation_identity_key, metadata.provider),
  concat_ws(' ', invocations.id, invocations.automation_id, invocations.idempotency_key, invocations.event_id, actions.context_id, metadata.project_id),
  invocations.created_at
FROM automate.automation_invocations AS invocations
INNER JOIN automate.actions AS actions ON actions.id = invocations.idempotency_key
INNER JOIN context_metadata AS metadata ON metadata.context_id = actions.context_id
UNION ALL
SELECT
  'contexts:' || contexts.id,
  'contexts',
  contexts.id,
  contexts.id,
  metadata.project_id,
  metadata.automation_id,
  metadata.provider,
  concat_ws(' ', 'context', contexts.status),
  concat_ws(E'\n', contexts.status, contexts.error::text),
  concat_ws(' ', metadata.project_name, metadata.automation_identity_key, metadata.provider),
  concat_ws(' ', contexts.id, metadata.automation_id, metadata.project_id),
  contexts.created_at
FROM automate.contexts
INNER JOIN context_metadata AS metadata ON metadata.context_id = contexts.id
UNION ALL
SELECT
  concat_ws(':', 'events', events.id, coalesce(event_links.id, 'unlinked')),
  'events',
  events.id,
  event_links.context_id,
  metadata.project_id,
  metadata.automation_id,
  split_part(events.event_type, '.', 1),
  events.event_type,
  concat_ws(E'\n', events.event_type, events.payload::text),
  concat_ws(' ', metadata.project_name, metadata.automation_identity_key, split_part(events.event_type, '.', 1)),
  concat_ws(' ', events.id, events.event_source_id, event_links.id, event_links.event_subscription_id, event_links.context_id, metadata.project_id, metadata.automation_id),
  events.created_at
FROM automate.events
LEFT JOIN automate.event_links ON event_links.event_id = events.id
LEFT JOIN context_metadata AS metadata ON metadata.context_id = event_links.context_id
UNION ALL
SELECT
  'actions:' || actions.id,
  'actions',
  actions.id,
  actions.context_id,
  metadata.project_id,
  metadata.automation_id,
  metadata.provider,
  concat_ws(' ', actions.name, actions.description),
  concat_ws(E'\n', actions.description, actions.status, actions.skip_reason, actions.error::text, actions.output::text),
  concat_ws(' ', metadata.project_name, metadata.automation_identity_key, metadata.provider),
  concat_ws(' ', actions.id, actions.context_id, metadata.project_id, metadata.automation_id, array_to_string(actions.action_dependency_ids, ' '), array_to_string(actions.event_dependency_ids, ' '), array_to_string(actions.signal_dependency_ids, ' ')),
  actions.created_at
FROM automate.actions
INNER JOIN context_metadata AS metadata ON metadata.context_id = actions.context_id
UNION ALL
SELECT
  'logs:' || logs.id,
  'logs',
  logs.id,
  logs.context_id,
  metadata.project_id,
  metadata.automation_id,
  metadata.provider,
  logs.message,
  concat_ws(E'\n', logs.message, logs.level, logs.fields::text),
  concat_ws(' ', metadata.project_name, metadata.automation_identity_key, metadata.provider),
  concat_ws(' ', logs.id, logs.context_id, logs.action_invocation_id, metadata.project_id, metadata.automation_id),
  logs.created_at
FROM automate.logs
INNER JOIN context_metadata AS metadata ON metadata.context_id = logs.context_id
UNION ALL
SELECT
  'outputs:' || outputs.id,
  'outputs',
  outputs.id,
  outputs.context_id,
  metadata.project_id,
  metadata.automation_id,
  metadata.provider,
  concat_ws(' ', outputs.type, 'output'),
  concat_ws(E'\n', outputs.type, outputs.data::text, outputs.value::text),
  concat_ws(' ', metadata.project_name, metadata.automation_identity_key, metadata.provider),
  concat_ws(' ', outputs.id, outputs.context_id, outputs.action_invocation_id, metadata.project_id, metadata.automation_id),
  outputs.created_at
FROM automate.outputs
INNER JOIN context_metadata AS metadata ON metadata.context_id = outputs.context_id
UNION ALL
SELECT
  'signal-invocations:' || signals.id,
  'signal-invocations',
  signals.id,
  signals.context_id,
  metadata.project_id,
  metadata.automation_id,
  metadata.provider,
  concat_ws(' ', 'signal', signals.status),
  concat_ws(E'\n', signals.status, signals.error::text, signals.output::text),
  concat_ws(' ', metadata.project_name, metadata.automation_identity_key, metadata.provider),
  concat_ws(' ', signals.id, signals.context_id, metadata.project_id, metadata.automation_id, array_to_string(signals.action_dependency_ids, ' '), array_to_string(signals.event_dependency_ids, ' '), array_to_string(signals.signal_dependency_ids, ' ')),
  signals.created_at
FROM automate.signal_invocations AS signals
INNER JOIN context_metadata AS metadata ON metadata.context_id = signals.context_id
UNION ALL
SELECT
  'signal-offers:' || offers.id,
  'signal-offers',
  offers.id,
  offers.context_id,
  metadata.project_id,
  metadata.automation_id,
  metadata.provider,
  concat_ws(' ', offers.lane, offers.status),
  concat_ws(E'\n', offers.lane, offers.status, offers.key::text),
  concat_ws(' ', metadata.project_name, metadata.automation_identity_key, metadata.provider),
  concat_ws(' ', offers.id, offers.context_id, offers.coordinator_id, offers.decision_id, offers.key_hash, metadata.project_id, metadata.automation_id, array_to_string(offers.action_dependency_ids, ' '), array_to_string(offers.event_dependency_ids, ' '), array_to_string(offers.signal_dependency_ids, ' ')),
  offers.created_at
FROM automate.signal_offers AS offers
INNER JOIN context_metadata AS metadata ON metadata.context_id = offers.context_id
), upserted AS (
INSERT INTO automate.search_documents AS current (
  document_id,
  resource,
  resource_id,
  context_id,
  project_id,
  automation_id,
  provider,
  title,
  content,
  metadata,
  identifiers,
  created_at
)
SELECT * FROM documents
WHERE $1::text IS NULL OR resource = $1
ON CONFLICT (document_id) DO UPDATE SET
  resource = excluded.resource,
  resource_id = excluded.resource_id,
  context_id = excluded.context_id,
  project_id = excluded.project_id,
  automation_id = excluded.automation_id,
  provider = excluded.provider,
  title = excluded.title,
  content = excluded.content,
  metadata = excluded.metadata,
  identifiers = excluded.identifiers,
  created_at = excluded.created_at
WHERE ROW(
  current.resource,
  current.resource_id,
  current.context_id,
  current.project_id,
  current.automation_id,
  current.provider,
  current.title,
  current.content,
  current.metadata,
  current.identifiers,
  current.created_at
) IS DISTINCT FROM ROW(
  excluded.resource,
  excluded.resource_id,
  excluded.context_id,
  excluded.project_id,
  excluded.automation_id,
  excluded.provider,
  excluded.title,
  excluded.content,
  excluded.metadata,
  excluded.identifiers,
  excluded.created_at
)
)
DELETE FROM automate.search_documents AS current
WHERE ($1::text IS NULL OR current.resource = $1)
AND NOT EXISTS (
  SELECT 1
  FROM documents
  WHERE ($1::text IS NULL OR documents.resource = $1)
  AND documents.document_id = current.document_id
);
`

export interface ExecutionDataSyncOptions {
  organizationId?: string
  projectId?: string
  since: string | null
}

export interface ExecutionDataSearchOptions {
  limit?: number
  query: string
}

export interface ExecutionDataSearchResult {
  content: string
  context: string | null
  id: string
  rank: number
  resource: string
  timestamp: string | null
}

export interface ExecutionDataStatus {
  consistency: "eventual"
  databasePath: string
  profileId: string
  resources: Record<
    LocalExecutionDatasetResource,
    { offset: string; rows: number; syncedAt: string }
  >
  schemaVersion: number
  scope: ExecutionDatasetScope
  syncedAt: string
}

export interface ExecutionDataSyncResult extends ExecutionDataStatus {
  recovery: typeof EXECUTION_DATA_RECOVERY | null
}

/** Actionable error returned after an unusable local cache is removed. */
class ExecutionDataCorruptionError extends Error {
  readonly code = "EXECUTION_DATA_CORRUPTED"

  /**
   * Creates the stable product-level replacement for a PostgreSQL error.
   *
   * @param cause - Original local corruption error.
   */
  constructor(cause: unknown) {
    super(
      "The corrupted local execution database was removed. Run `automate data sync` to rebuild it. Production execution data was not changed.",
      { cause },
    )
  }
}

/**
 * Parses an ISO timestamp or compact lookback such as `7d`.
 *
 * @param value - User-supplied absolute time, lookback, or `all`.
 * @throws When the value is not a supported time expression.
 */
export function parseExecutionDataSince(value: string) {
  if (value === "all") return null

  const duration = /^(\d+)(m|h|d|w)$/.exec(value)
  if (duration) {
    const unit = duration[2]
    return new Date(
      Date.now() -
        Number(duration[1]) *
          (unit === "m"
            ? 60_000
            : unit === "h"
              ? 3_600_000
              : unit === "d"
                ? 86_400_000
                : 604_800_000),
    ).toISOString()
  }

  const date = new Date(value)
  if (Number.isNaN(date.getTime())) {
    throw new Error(
      "Since must be an ISO timestamp, a lookback like 7d, or all.",
    )
  }
  return date.toISOString()
}

/**
 * Synchronizes one authorized scope into its isolated local database.
 *
 * @param options - Organization, project, and history scope.
 */
export async function syncExecutionData(options: ExecutionDataSyncOptions) {
  const manifest = await fetchManifest(options)
  return await useProfileDatabase(
    manifest.profileId,
    { rebuildOnCorruption: true, resetOnVersionMismatch: true },
    async (database, recoveredFromCorruption) => {
      const abortController = new AbortController()
      const shapes = createShapes(manifest, false, abortController.signal)

      try {
        await replaceDataset(
          database,
          manifest,
          await Promise.all(
            shapes.map(async ({ name, shape }) => {
              return getResourceSnapshot(name, shape, await shape.rows)
            }),
          ),
        )
        await rememberProfile(manifest.profileId)
        return {
          ...(await getStatus(database, manifest.profileId)),
          recovery: recoveredFromCorruption ? EXECUTION_DATA_RECOVERY : null,
        } satisfies ExecutionDataSyncResult
      } finally {
        abortController.abort()
        shapes.forEach(({ shape }) => shape.stream.unsubscribeAll())
      }
    },
  )
}

/**
 * Keeps an authorized scope synchronized until the supplied signal aborts.
 *
 * @param options - Organization, project, and history scope.
 * @param signal - Cancellation signal for the live stream.
 * @param onChange - Optional callback after a materialized update.
 */
export async function followExecutionData(
  options: ExecutionDataSyncOptions,
  signal: AbortSignal,
  onChange?: () => Promise<void>,
) {
  let manifest: ExecutionDatasetManifest
  try {
    manifest = await fetchManifest(options, signal)
  } catch (error) {
    if (signal.aborted) return
    throw error
  }
  const shapes = createShapes(manifest, true, signal)
  let writeChain = Promise.resolve()
  let rejectFollow: ((error: unknown) => void) | undefined
  const followFailed = new Promise<never>((_, reject) => {
    rejectFollow = reject
  })

  try {
    try {
      const snapshots = await Promise.all(
        shapes.map(async ({ name, shape }) => {
          return getResourceSnapshot(name, shape, await shape.rows)
        }),
      )
      await useProfileDatabase(
        manifest.profileId,
        {
          rebuildOnCorruption: true,
          resetOnVersionMismatch: true,
          signal,
        },
        async (database) => {
          await replaceDataset(database, manifest, snapshots)
          await rememberProfile(manifest.profileId)
        },
      )
    } catch (error) {
      if (signal.aborted) return
      throw error
    }
    await onChange?.()

    const unsubscribers = shapes.map(({ name, shape }) =>
      shape.subscribe(({ rows: nextRows }) => {
        let resource: ResourceSnapshot
        try {
          resource = getResourceSnapshot(name, shape, nextRows)
        } catch (error) {
          rejectFollow?.(error)
          return
        }
        writeChain = writeChain
          .then(async () => {
            if (signal.aborted) return
            await useProfileDatabase(
              manifest.profileId,
              {
                rebuildOnCorruption: false,
                resetOnVersionMismatch: true,
                signal,
              },
              async (database) => {
                await replaceResource(database, manifest, resource)
              },
            )
            await onChange?.()
          })
          .catch((error: unknown) => {
            rejectFollow?.(error)
          })
      }),
    )

    try {
      await Promise.race([
        new Promise<void>((resolve) => {
          if (signal.aborted) resolve()
          else signal.addEventListener("abort", () => resolve(), { once: true })
        }),
        followFailed,
      ])
      await writeChain
    } finally {
      unsubscribers.forEach((unsubscribe) => unsubscribe())
      shapes.forEach(({ shape }) => shape.stream.unsubscribeAll())
    }
  } finally {
    shapes.forEach(({ shape }) => shape.stream.unsubscribeAll())
  }
}

/**
 * Executes SQL as a restricted local role with read-only dataset access.
 *
 * @param sql - One or more PostgreSQL statements.
 */
export async function queryExecutionData(sql: string) {
  return await useActiveDatabase(async (database, profileId) => {
    return {
      profileId,
      results: normalizeQueryResults(
        await executeReadOnlyExecutionDataQuery(database, sql),
      ),
    }
  })
}

/**
 * Searches the active local execution dataset by relevance.
 *
 * @param options - Search text and optional result limit.
 */
export async function searchExecutionData(options: ExecutionDataSearchOptions) {
  return await useActiveDatabase(async (database, profileId) => {
    return {
      profileId,
      query: options.query,
      results: await searchExecutionDataDocuments(database, options),
    }
  })
}

/** Returns metadata and synchronized row counts for the active profile. */
export async function getExecutionDataStatus() {
  return await useActiveDatabase(async (database, profileId) => {
    return await getStatus(database, profileId)
  })
}

/**
 * Exports the complete active PGlite database as a compressed archive.
 *
 * @param outputPath - Destination path for the archive.
 */
export async function exportExecutionData(outputPath: string) {
  return await useActiveDatabase(async (database, profileId) => {
    const archive = await database.dumpDataDir("gzip")
    await atomicWritePrivateFile(
      outputPath,
      new Uint8Array(await archive.arrayBuffer()),
    )
    return { outputPath, profileId, size: archive.size }
  })
}

/** Permanently removes the active profile's local execution database. */
export async function clearExecutionData() {
  return await withExecutionDataLock(async () => {
    const profileId = await readActiveProfile()
    if (!profileId || loadExecutionDataProfileSync() !== profileId) {
      return { cleared: false, profileId: null }
    }

    await removeExecutionDataProfile(profileId)
    return { cleared: true, profileId }
  })
}

/** Removes every local execution-data profile for the configured origin. */
export async function clearAllExecutionData() {
  await withExecutionDataLock(async () => {
    await fs.rm(getOriginDirectory(), { force: true, recursive: true })
  })
}

/** Returns the active local database path without opening it. */
export async function getActiveExecutionDataPath() {
  return await withExecutionDataLock(async () => {
    const profileId = await readActiveProfile()
    return profileId ? getDatabasePath(profileId) : null
  })
}

/**
 * Fetches and validates the server-owned dataset manifest.
 *
 * @param options - Organization, project, and history scope.
 * @param signal - Optional request cancellation signal.
 */
async function fetchManifest(
  options: ExecutionDataSyncOptions,
  signal?: AbortSignal,
) {
  if (!loadSavedTokenSync()) {
    throw new Error(
      "Execution data requires a logged-in user session. Run automate login first.",
    )
  }
  return await sessionApiClient.executionData.manifest(
    {
      organizationId: options.organizationId,
      projectId: options.projectId,
      since: options.since ?? undefined,
    },
    { signal },
  )
}

/**
 * Creates one authenticated Electric shape for each manifest resource.
 *
 * @param manifest - Authorized resource URLs and scope.
 * @param subscribe - Whether streams should remain live.
 * @param signal - Optional cancellation signal for initial and live loading.
 * @throws When the saved login disappeared after loading the manifest.
 */
function createShapes(
  manifest: ExecutionDatasetManifest,
  subscribe: boolean,
  signal?: AbortSignal,
) {
  const token = loadSavedTokenSync()
  if (!token) throw new Error("The saved CLI session is unavailable.")

  return manifest.resources.map(({ name, url }) => ({
    name,
    shape: new Shape(
      new ShapeStream<ElectricRow>({
        url,
        headers: { authorization: `Bearer ${token}` },
        liveSse: subscribe && url.startsWith("https:"),
        parser: { bytea: parseBytea },
        signal,
        subscribe,
        warnOnHttp: false,
      }),
    ),
  }))
}

/**
 * Serializes every PGlite owner for the configured application origin.
 *
 * @param operation - Filesystem or database work requiring exclusive access.
 * @param signal - Optional cancellation while waiting for another owner.
 */
async function withExecutionDataLock<T>(
  operation: () => Promise<T>,
  signal?: AbortSignal,
) {
  await fs.mkdir(EXECUTION_DATA_ROOT, { mode: 0o700, recursive: true })
  await fs.chmod(EXECUTION_DATA_ROOT, 0o700)

  while (true) {
    signal?.throwIfAborted()
    let release: (() => Promise<void>) | undefined
    try {
      release = await lock(getOriginDirectory(), {
        realpath: false,
        stale: EXECUTION_DATA_LOCK_STALE_MS,
        update: EXECUTION_DATA_LOCK_UPDATE_MS,
      })
    } catch (error) {
      if (!isExecutionDataLockedError(error)) throw error
      await setTimeout(EXECUTION_DATA_LOCK_RETRY_MS, undefined, { signal })
      continue
    }

    try {
      return await operation()
    } finally {
      await release()
    }
  }
}

/**
 * Returns whether another process currently owns the execution-data lock.
 *
 * @param error - Unknown lock acquisition failure.
 */
function isExecutionDataLockedError(error: unknown) {
  return (
    typeof error === "object" &&
    error !== null &&
    Reflect.get(error, "code") === "ELOCKED"
  )
}

/**
 * Opens and initializes one profile database with owner-only storage.
 *
 * @param profileId - Opaque authorized profile identity.
 * @param resetOnVersionMismatch - Whether synchronization may replace stale
 *   managed tables.
 */
async function openDatabase(profileId: string, resetOnVersionMismatch = false) {
  const directory = getProfileDirectory(profileId)
  await fs.mkdir(directory, { mode: 0o700, recursive: true })
  await fs.chmod(directory, 0o700)
  const database = await PGlite.create(getDatabasePath(profileId), {
    extensions: { pg_trgm },
  })
  try {
    await ensureExecutionDataSchema(database, resetOnVersionMismatch)
    return database
  } catch (error) {
    await database.close()
    throw error
  }
}

interface ProfileDatabaseOptions {
  rebuildOnCorruption: boolean
  resetOnVersionMismatch: boolean
  signal?: AbortSignal
}

/**
 * Runs one local database operation and handles explicit PostgreSQL corruption.
 *
 * @param profileId - Opaque authorized profile identity.
 * @param options - Schema and one-shot rebuild behavior.
 * @param operation - Work to perform against the opened local database.
 */
export async function useProfileDatabase<T>(
  profileId: string,
  options: ProfileDatabaseOptions,
  operation: (database: PGlite, recoveredFromCorruption: boolean) => Promise<T>,
) {
  return await withExecutionDataLock(
    async () => await useProfileDatabaseUnlocked(profileId, options, operation),
    options.signal,
  )
}

/**
 * Runs one local database operation after its origin lock is held.
 *
 * @param profileId - Opaque authorized profile identity.
 * @param options - Schema and one-shot rebuild behavior.
 * @param operation - Work to perform against the opened local database.
 */
async function useProfileDatabaseUnlocked<T>(
  profileId: string,
  options: ProfileDatabaseOptions,
  operation: (database: PGlite, recoveredFromCorruption: boolean) => Promise<T>,
) {
  let recoveredFromCorruption = false

  while (true) {
    let database: PGlite | undefined
    try {
      database = await openDatabase(profileId, options.resetOnVersionMismatch)
      // Preserve the operation result until the database flushes successfully.
      const result = await operation(database, recoveredFromCorruption)
      // Detach before closing so a close failure is not closed a second time.
      const databaseToClose = database
      database = undefined
      await databaseToClose.close()
      return result
    } catch (error) {
      await database?.close().catch(() => undefined)
      if (!isExecutionDataCorruptionError(error)) throw error

      await removeExecutionDataProfile(profileId)
      if (options.rebuildOnCorruption && !recoveredFromCorruption) {
        recoveredFromCorruption = true
        continue
      }
      throw new ExecutionDataCorruptionError(error)
    }
  }
}

/**
 * Returns whether an error is an explicit PostgreSQL corruption condition.
 *
 * @param error - Unknown operation failure.
 */
export function isExecutionDataCorruptionError(error: unknown) {
  if (typeof error !== "object" || error === null) return false
  return EXECUTION_DATA_CORRUPTION_CODES.has(String(Reflect.get(error, "code")))
}

/**
 * Initializes or intentionally replaces the managed local query schema.
 *
 * @param database - Local PGlite database to initialize.
 * @param resetOnVersionMismatch - Whether incompatible managed tables may be
 *   replaced.
 */
export async function ensureExecutionDataSchema(
  database: PGlite,
  resetOnVersionMismatch = false,
) {
  await database.exec(EXECUTION_DATA_BOOTSTRAP_SQL)
  const localVersion = (
    await database.query<{ schema_version: number }>(
      "SELECT schema_version FROM automate_internal.sync_state WHERE singleton",
    )
  ).rows[0]?.schema_version
  const hasLegacyTraceSchema = (
    await database.query<{ exists: boolean }>(`
      SELECT
        to_regclass('automate.trace_spans') IS NOT NULL
        OR to_regclass('automate.trace_span_ancestors') IS NOT NULL
        OR to_regclass('automate.trace_span_descendants') IS NOT NULL
        OR EXISTS (
          SELECT 1
          FROM information_schema.columns
          WHERE table_schema = 'automate'
            AND table_name = 'logs'
            AND column_name = 'span_index'
        ) AS exists
    `)
  ).rows[0]?.exists
  if (
    (localVersion !== undefined &&
      localVersion !== LOCAL_EXECUTION_DATASET_VERSION) ||
    hasLegacyTraceSchema
  ) {
    if (!resetOnVersionMismatch) {
      throw new Error(
        localVersion === undefined
          ? "The local execution dataset contains an obsolete trace schema; run automate data sync to replace it."
          : `The local execution dataset uses schema version ${localVersion}; run automate data sync to replace it with version ${LOCAL_EXECUTION_DATASET_VERSION}.`,
      )
    }
    await database.exec(`
      DROP SCHEMA IF EXISTS automate CASCADE;
      DELETE FROM automate_internal.resource_sync_state;
      DELETE FROM automate_internal.sync_state;
      DELETE FROM automate_internal.local_state;
    `)
  }
  const searchSchemaMismatch =
    (
      await database.query<{ search_schema_version: number }>(
        "SELECT search_schema_version FROM automate_internal.local_state WHERE singleton",
      )
    ).rows[0]?.search_schema_version !== EXECUTION_DATA_SEARCH_SCHEMA_VERSION
  if (searchSchemaMismatch) {
    await database.exec(`
      CREATE SCHEMA IF NOT EXISTS automate;
      DROP TABLE IF EXISTS automate.search_documents CASCADE;
      DELETE FROM automate_internal.local_state;
    `)
  }
  await database.exec(EXECUTION_DATA_SCHEMA_SQL)
  if (searchSchemaMismatch) {
    await database.transaction(async (transaction) => {
      await rebuildExecutionDataSearchDocuments(transaction)
      await writeSearchSchemaState(transaction)
    })
  }
}

/**
 * Runs an operation against the database authorized for the current login.
 *
 * @param operation - Work to perform against the active local database.
 */
async function useActiveDatabase<T>(
  operation: (database: PGlite, profileId: string) => Promise<T>,
) {
  return await withExecutionDataLock(async () => {
    const profileId = await readActiveProfile()
    if (!profileId) {
      throw new Error(
        "No local execution dataset. Run automate data sync first.",
      )
    }
    if (loadExecutionDataProfileSync() !== profileId) {
      throw new Error(
        "The local execution dataset belongs to another login session. Run automate data sync first.",
      )
    }
    return await useProfileDatabaseUnlocked(
      profileId,
      { rebuildOnCorruption: false, resetOnVersionMismatch: false },
      async (database) => await operation(database, profileId),
    )
  })
}

interface ResourceSnapshot {
  name: LocalExecutionDatasetResource
  offset: string
  rows: ElectricRow[]
  syncedAt: string
}

/**
 * Captures one shape's independent freshness boundary with its rows.
 *
 * @param name - Resource represented by the Shape.
 * @param shape - Materialized Electric Shape.
 * @param rows - Current complete Shape rows.
 * @throws When Electric has not published an up-to-date boundary.
 */
function getResourceSnapshot(
  name: LocalExecutionDatasetResource,
  shape: Shape<ElectricRow>,
  rows: ElectricRow[],
): ResourceSnapshot {
  const syncedAt = shape.lastSyncedAt()
  if (!shape.isUpToDate || syncedAt === undefined) {
    throw new Error(`Electric did not finish synchronizing ${name}.`)
  }
  return {
    name,
    offset: shape.lastOffset,
    rows,
    syncedAt: new Date(syncedAt).toISOString(),
  }
}

/**
 * Atomically replaces local tables with independently current shape snapshots.
 *
 * @param database - Active local PGlite database.
 * @param manifest - Authorized scope recorded with the snapshot.
 * @param resources - Complete initial resource snapshots.
 */
export async function replaceDataset(
  database: PGlite,
  manifest: ExecutionDatasetManifest,
  resources: readonly ResourceSnapshot[],
) {
  const preparedResources = await Promise.all(
    resources.map(async ({ name, offset, rows, syncedAt }) => ({
      name,
      offset,
      rows: await prepareRows(name, rows),
      syncedAt,
    })),
  )

  const snapshots = new Map(
    preparedResources.map((resource) => [resource.name, resource]),
  )
  if (snapshots.size !== preparedResources.length) {
    throw new Error("Execution-data snapshots must have unique resources.")
  }

  await database.transaction(async (transaction) => {
    for (const name of LOCAL_EXECUTION_DATASET_RESOURCE_SCHEMA.options) {
      await reconcileRows(transaction, name, snapshots.get(name)?.rows ?? [])
    }
    await deleteStaleResourceSyncStates(transaction, [...snapshots.keys()])
    for (const { name, offset, syncedAt } of preparedResources) {
      await writeResourceSyncState(transaction, name, offset, syncedAt)
    }
    await rebuildExecutionDataSearchDocuments(transaction)
    await writeSearchSchemaState(transaction)
    await writeSyncState(transaction, manifest)
  })
}

/**
 * Replaces one live resource after Electric publishes an update.
 *
 * @param database - Active local PGlite database.
 * @param manifest - Authorized scope recorded with the update.
 * @param resource - Complete resource rows and independent freshness boundary.
 */
export async function replaceResource(
  database: PGlite,
  manifest: ExecutionDatasetManifest,
  resource: ResourceSnapshot,
) {
  const preparedRows = await prepareRows(resource.name, resource.rows)
  await database.transaction(async (transaction) => {
    await reconcileRows(transaction, resource.name, preparedRows)
    await writeResourceSyncState(
      transaction,
      resource.name,
      resource.offset,
      resource.syncedAt,
    )
    const searchResource = getIncrementalSearchResource(resource.name)
    if (searchResource !== null) {
      await rebuildExecutionDataSearchDocuments(transaction, searchResource)
    }
    await writeSearchSchemaState(transaction)
    await writeSyncState(transaction, manifest)
  })
}

/**
 * Converts one resource snapshot to local SQL rows.
 *
 * @param resource - Resource being materialized.
 * @param rows - Electric snapshot rows.
 */
async function prepareRows(
  resource: LocalExecutionDatasetResource,
  rows: ElectricRow[],
) {
  return await Promise.all(rows.map((row) => prepareRow(resource, row)))
}

/**
 * Decodes and projects payload columns for one synchronized row.
 *
 * @param resource - Resource owning the row.
 * @param row - Raw Electric row.
 */
async function prepareRow(
  resource: LocalExecutionDatasetResource,
  row: ElectricRow,
) {
  switch (resource) {
    case "action-attempts":
      return { ...row, error: stringifyJson(row.error) }
    case "actions": {
      const bytes = optionalBytes(row.output)
      return {
        ...row,
        error: stringifyJson(row.error),
        output_sensitivity: stringifyJson(row.output_sensitivity),
        output: bytes
          ? stringifyJson(
              await projectExecutionDataValue(
                await decode(bytes),
                `automate://actions/${requireString(row.id, "action ID")}/output`,
              ),
            )
          : null,
        output_encoded: bytes,
      }
    }
    case "events": {
      const bytes = requireBytes(row.payload, "event payload")
      return {
        ...row,
        payload: stringifyJson(
          await projectExecutionDataValue(
            await decode(bytes),
            `automate://events/${requireString(row.id, "event ID")}/payload`,
          ),
        ),
        payload_encoded: bytes,
      }
    }
    case "logs": {
      const bytes = optionalBytes(row.fields)
      return {
        ...row,
        fields: bytes
          ? stringifyJson(
              await projectExecutionDataValue(
                await decode(bytes),
                `automate://logs/${requireString(row.id, "log ID")}/fields`,
              ),
            )
          : null,
        fields_encoded: bytes,
        sensitivity: stringifyJson(row.sensitivity),
      }
    }
    case "outputs": {
      const bytes = requireBytes(row.value, "automation output")
      const id = requireString(row.id, "output ID")
      const value = await decode(bytes)
      const record =
        value &&
        typeof value === "object" &&
        !Array.isArray(value) &&
        !(value instanceof Map) &&
        !(value instanceof Set)
          ? value
          : null
      const type = record && "type" in record ? record.type : null
      return {
        ...row,
        data: stringifyJson(
          await projectExecutionDataValue(
            record && "data" in record ? record.data : undefined,
            `automate://outputs/${id}/data`,
          ),
        ),
        type: typeof type === "string" ? type : null,
        value: stringifyJson(
          await projectExecutionDataValue(
            value,
            `automate://outputs/${id}/value`,
          ),
        ),
        value_encoded: bytes,
        value_sensitivity: stringifyJson(row.value_sensitivity),
      }
    }
    case "signal-invocations": {
      const bytes = optionalBytes(row.output)
      return {
        ...row,
        error: stringifyJson(row.error),
        output: bytes
          ? stringifyJson(
              await projectExecutionDataValue(
                await decode(bytes),
                `automate://signal-invocations/${requireString(row.id, "signal invocation ID")}/output`,
              ),
            )
          : null,
        output_encoded: bytes,
      }
    }
    case "signal-offers": {
      const bytes = requireBytes(row.key, "signal offer key")
      return {
        ...row,
        key: stringifyJson(
          await projectExecutionDataValue(
            await decode(bytes),
            `automate://signal-offers/${requireString(row.id, "signal offer ID")}/key`,
          ),
        ),
        key_encoded: bytes,
      }
    }
    case "signal-partitions": {
      const bytes = requireBytes(row.key, "signal partition key")
      return {
        ...row,
        key: stringifyJson(
          await projectExecutionDataValue(
            await decode(bytes),
            `automate://signal-partitions/${requireString(row.id, "signal partition ID")}/key`,
          ),
        ),
        key_encoded: bytes,
        state: stringifyJson(row.state),
      }
    }
    case "contexts":
      return { ...row, error: stringifyJson(row.error) }
    case "event-links":
      return { ...row, config: stringifyJson(row.config) }
    default:
      return row
  }
}

/**
 * Reconciles prepared rows through fixed table, column, and key definitions.
 *
 * @param database - Local query executor or transaction.
 * @param resource - Destination dataset resource.
 * @param rows - Prepared local rows.
 */
async function reconcileRows(
  database: Pick<PGlite, "query">,
  resource: LocalExecutionDatasetResource,
  rows: Record<string, unknown>[],
) {
  const definition = RESOURCE_DEFINITIONS[resource]
  const primaryKeys: readonly string[] = RESOURCE_PRIMARY_KEYS[resource]
  const updateColumns = definition.columns.filter(
    (column) => !primaryKeys.includes(column),
  )
  const jsonColumns = new Set<string>(
    "jsonColumns" in definition ? definition.jsonColumns : [],
  )
  const sql = `INSERT INTO automate.${definition.table} AS current (${definition.columns.join(", ")}) VALUES (${definition.columns.map((column, index) => (jsonColumns.has(column) ? `$${index + 1}::jsonb` : `$${index + 1}`)).join(", ")})
    ON CONFLICT (${primaryKeys.join(", ")}) DO UPDATE SET
      ${updateColumns.map((column) => `${column} = excluded.${column}`).join(", ")}
    WHERE ROW(${updateColumns.map((column) => `current.${column}`).join(", ")})
      IS DISTINCT FROM ROW(${updateColumns.map((column) => `excluded.${column}`).join(", ")})`
  const identities = rows.map((row) =>
    Object.fromEntries(
      primaryKeys.map((column) => [
        column,
        requireString(row[column], `${resource} ${column}`),
      ]),
    ),
  )
  if (
    new Set(
      identities.map((identity) =>
        primaryKeys.map((column) => identity[column]).join("\0"),
      ),
    ).size !== identities.length
  ) {
    throw new Error(`${resource} snapshot rows must have unique identities.`)
  }

  for (const row of rows) {
    await database.query(
      sql,
      definition.columns.map((column) => row[column] ?? null),
    )
  }
  await database.query(
    `DELETE FROM automate.${definition.table} AS current
    WHERE NOT EXISTS (
      SELECT 1
      FROM jsonb_to_recordset($1::jsonb) AS incoming(
        ${primaryKeys.map((column) => `${column} text`).join(", ")}
      )
      WHERE ${primaryKeys.map((column) => `incoming.${column} = current.${column}`).join(" AND ")}
    )`,
    [JSON.stringify(identities)],
  )
}

/**
 * Rebuilds the full local corpus or one independently searchable resource.
 *
 * @param database - Local query executor or transaction.
 * @param resource - Optional search resource to replace incrementally.
 */
export async function rebuildExecutionDataSearchDocuments(
  database: Pick<PGlite, "query">,
  resource?: LocalExecutionDatasetResource,
) {
  await database.query(EXECUTION_DATA_SEARCH_INSERT_SQL, [
    resource === "actions" || resource === "automation-invocations"
      ? null
      : (resource ?? null),
  ])
}

/**
 * Maps a synchronized resource to the derived documents it can refresh alone.
 *
 * `undefined` requests a full rebuild, while `null` means the resource has no
 * searchable projection.
 *
 * @param resource - Synchronized execution-data resource.
 */
function getIncrementalSearchResource(
  resource: LocalExecutionDatasetResource,
): LocalExecutionDatasetResource | null | undefined {
  switch (resource) {
    case "actions":
    case "automation-invocations":
    case "event-links":
    case "events":
      return undefined
    case "logs":
    case "outputs":
    case "signal-invocations":
    case "signal-offers":
      return resource
    case "action-attempts":
      return null
    case "automations":
    case "contexts":
    case "projects":
      return undefined
    default:
      return null
  }
}

/**
 * Records the derived search schema currently materialized in the database.
 *
 * @param database - Local query executor or transaction.
 */
async function writeSearchSchemaState(database: Pick<PGlite, "query">) {
  await database.query(
    `INSERT INTO automate_internal.local_state AS current (
      singleton, search_schema_version
    ) VALUES (true, $1)
    ON CONFLICT (singleton) DO UPDATE SET
      search_schema_version = excluded.search_schema_version
    WHERE current.search_schema_version
      IS DISTINCT FROM excluded.search_schema_version`,
    [EXECUTION_DATA_SEARCH_SCHEMA_VERSION],
  )
}

/**
 * Records the independent Electric boundary materialized for one resource.
 *
 * @param database - Local query executor or transaction.
 * @param resource - Resource whose boundary was materialized.
 * @param offset - Shape-log offset represented by the local rows.
 * @param syncedAt - Time Electric marked the Shape up to date.
 */
async function writeResourceSyncState(
  database: Pick<PGlite, "query">,
  resource: LocalExecutionDatasetResource,
  offset: string,
  syncedAt: string,
) {
  await database.query(
    `INSERT INTO automate_internal.resource_sync_state AS current (
      resource, shape_offset, synced_at
    ) VALUES ($1, $2, $3)
    ON CONFLICT (resource) DO UPDATE SET
      shape_offset = excluded.shape_offset,
      synced_at = excluded.synced_at
    WHERE ROW(current.shape_offset, current.synced_at)
      IS DISTINCT FROM ROW(excluded.shape_offset, excluded.synced_at)`,
    [resource, offset, syncedAt],
  )
}

/**
 * Removes freshness boundaries for resources absent from a full snapshot.
 *
 * @param database - Local query executor or transaction.
 * @param resources - Resources represented by the complete snapshot.
 */
async function deleteStaleResourceSyncStates(
  database: Pick<PGlite, "query">,
  resources: LocalExecutionDatasetResource[],
) {
  await database.query(
    `DELETE FROM automate_internal.resource_sync_state
    WHERE NOT (resource = ANY($1::text[]))`,
    [resources],
  )
}

/**
 * Writes the active scope and synchronization timestamp.
 *
 * @param database - Local query executor or transaction.
 * @param manifest - Authorized scope to record.
 */
async function writeSyncState(
  database: Pick<PGlite, "query">,
  manifest: ExecutionDatasetManifest,
) {
  await database.query(
    `INSERT INTO automate_internal.sync_state (
      singleton, profile_id, schema_version, organization_id, project_id, since_at, synced_at
    ) VALUES (true, $1, $2, $3, $4, $5, now())
    ON CONFLICT (singleton) DO UPDATE SET
      profile_id = excluded.profile_id,
      schema_version = excluded.schema_version,
      organization_id = excluded.organization_id,
      project_id = excluded.project_id,
      since_at = excluded.since_at,
      synced_at = excluded.synced_at`,
    [
      manifest.profileId,
      LOCAL_EXECUTION_DATASET_VERSION,
      manifest.scope.organizationId,
      manifest.scope.projectId,
      manifest.scope.since,
    ],
  )
}

/**
 * Reads local scope metadata and per-resource counts.
 *
 * @param database - Active local PGlite database.
 * @param profileId - Profile whose database path is reported.
 */
async function getStatus(database: PGlite, profileId: string) {
  const row = (
    await database.query<{
      organization_id: string | null
      profile_id: string
      project_id: string | null
      schema_version: number
      since_at: string | null
      synced_at: string
    }>("SELECT * FROM automate_internal.sync_state WHERE singleton")
  ).rows[0]
  if (!row) throw new Error("The local execution dataset has not been synced.")

  // Keep the remote boundary next to each count before validating the complete record.
  const resourceStates = new Map(
    await Promise.all(
      Object.entries(RESOURCE_DEFINITIONS).map(
        async ([name, definition]) =>
          [
            name,
            {
              rows: Number(
                (
                  await database.query<{ count: bigint }>(
                    `SELECT count(*) AS count FROM automate.${definition.table}`,
                  )
                ).rows[0]?.count ?? 0n,
              ),
              state: (
                await database.query<{
                  shape_offset: string
                  synced_at: string
                }>(
                  `SELECT shape_offset, synced_at
                  FROM automate_internal.resource_sync_state
                  WHERE resource = $1`,
                  [name],
                )
              ).rows[0],
            },
          ] as const,
      ),
    ),
  )
  // Validate that every versioned resource has a complete freshness entry.
  const resources = EXECUTION_DATA_RESOURCE_STATUS_SCHEMA.parse(
    Object.fromEntries(
      [...resourceStates].map(([name, { rows, state }]) => {
        if (!state) throw new Error(`Local ${name} freshness is unavailable.`)
        return [
          name,
          {
            offset: state.shape_offset,
            rows,
            syncedAt: new Date(state.synced_at).toISOString(),
          },
        ]
      }),
    ),
  )
  return {
    consistency: "eventual",
    databasePath: getDatabasePath(profileId),
    profileId: row.profile_id,
    resources,
    schemaVersion: row.schema_version,
    scope: {
      organizationId: row.organization_id,
      projectId: row.project_id,
      since: row.since_at ? new Date(row.since_at).toISOString() : null,
    },
    syncedAt: new Date(row.synced_at).toISOString(),
  } satisfies ExecutionDataStatus
}

/**
 * Executes only query statements inside a database-enforced read-only boundary.
 *
 * @param database - Initialized local execution-data database.
 * @param sql - One or more PostgreSQL query statements.
 */
export async function executeReadOnlyExecutionDataQuery(
  database: PGlite,
  sql: string,
) {
  await authorizeExecutionDataQuery(sql)
  return await database.transaction(async (transaction) => {
    await transaction.exec(`
      SET TRANSACTION READ ONLY;
      SET LOCAL ROLE automate_query;
      SET LOCAL search_path = automate, automate_extensions, pg_catalog;
    `)
    return await transaction.exec(sql)
  })
}

/**
 * Searches a prepared execution-data database with full-text and trigram
 * ranking.
 *
 * @param database - Initialized local execution-data database.
 * @param options - Search text and optional result limit.
 */
export async function searchExecutionDataDocuments(
  database: PGlite,
  options: ExecutionDataSearchOptions,
) {
  return (
    await database.transaction(async (transaction) => {
      await transaction.exec(`
        SET TRANSACTION READ ONLY;
        SET LOCAL ROLE automate_query;
        SET LOCAL search_path = automate, automate_extensions, pg_catalog;
      `)
      return (
        await transaction.query<{
          content: string
          context: string | null
          id: string
          rank: number
          resource: string
          timestamp: Date | string | null
        }>(
          `WITH input AS (
          SELECT
            $1::text AS term,
            websearch_to_tsquery('simple', $1) AS query,
            strpos($1, '"') = 0
              AND $1 !~ '(^|[[:space:]])-'
              AND $1 !~* '(^|[[:space:]])OR([[:space:]]|$)' AS allow_fuzzy
        ), ranked AS (
          SELECT
            documents.*,
            input.term,
            input.query,
            ts_rank_cd(
              ARRAY[0.05, 0.2, 0.5, 1.0]::real[],
              documents.search_vector,
              input.query,
              32
            ) AS text_rank,
            greatest(
              word_similarity(input.term, documents.title),
              word_similarity(input.term, documents.content),
              word_similarity(input.term, documents.identifiers)
            ) AS fuzzy_rank,
            CASE
              WHEN lower(documents.resource_id) = lower(input.term) THEN 1.0
              WHEN documents.identifiers ILIKE $2 ESCAPE '\\' THEN 0.5
              ELSE 0.0
            END AS identifier_rank
          FROM automate.search_documents AS documents
          CROSS JOIN input
          WHERE documents.search_vector @@ input.query
            OR input.allow_fuzzy AND (
              documents.identifiers ILIKE $2 ESCAPE '\\'
              OR input.term OPERATOR(automate_extensions.<%) documents.title
              OR input.term OPERATOR(automate_extensions.<%) documents.content
              OR input.term OPERATOR(automate_extensions.<%) documents.identifiers
            )
        )
        SELECT
          resource,
          resource_id AS id,
          context_id AS context,
          created_at AS timestamp,
          CASE
            WHEN search_vector @@ query THEN ts_headline(
              'simple',
              concat_ws(E'\\n', title, content, metadata, identifiers),
              query,
              'StartSel=[[ , StopSel= ]], MaxFragments=2, MaxWords=24, MinWords=8'
            )
            ELSE left(concat_ws(E'\\n', title, content), 500)
          END AS content,
          text_rank + fuzzy_rank * 0.35 + identifier_rank AS rank
        FROM ranked
        ORDER BY rank DESC, created_at DESC NULLS LAST, resource, resource_id
        LIMIT $3`,
          [
            options.query,
            `%${options.query.replaceAll("\\", "\\\\").replaceAll("%", "\\%").replaceAll("_", "\\_")}%`,
            options.limit ?? 20,
          ],
        )
      ).rows
    })
  ).map((row) => ({
    ...row,
    rank: row.rank,
    timestamp: row.timestamp ? new Date(row.timestamp).toISOString() : null,
  })) satisfies ExecutionDataSearchResult[]
}

/**
 * Rejects every statement or nested command that is not a pure query.
 *
 * @param sql - Complete PostgreSQL input to authorize.
 */
async function authorizeExecutionDataQuery(sql: string) {
  const statements = (await parse(sql)).stmts ?? []
  if (statements.length === 0) {
    throw new Error("The SQL query is empty.")
  }

  for (const { stmt } of statements) {
    if (!stmt) throw new Error("The SQL query contains an empty statement.")
    const statementType = Object.keys(stmt)[0]
    if (statementType !== "SelectStmt" && statementType !== "ExplainStmt") {
      throw new Error(
        `Execution data is read-only; ${formatStatementType(statementType)} is not allowed.`,
      )
    }
    assertReadOnlySqlNode(stmt)
  }
}

/**
 * Walks libpg_query's AST to reject side effects hidden inside query syntax.
 *
 * @param value - Current unknown AST node or scalar.
 * @throws When a nested statement or session mutation is not read-only.
 */
function assertReadOnlySqlNode(value: unknown): void {
  if (Array.isArray(value)) {
    value.forEach(assertReadOnlySqlNode)
    return
  }
  if (!value || typeof value !== "object") return

  for (const [key, child] of Object.entries(value)) {
    if (key.endsWith("Stmt") && key !== "SelectStmt" && key !== "ExplainStmt") {
      throw new Error(
        `Execution data is read-only; ${formatStatementType(key)} is not allowed.`,
      )
    }
    if ((key === "intoClause" || key === "lockingClause") && child) {
      throw new Error("Execution data queries cannot create or lock rows.")
    }
    if (key === "FuncCall" && isSessionMutationFunction(child)) {
      throw new Error(
        "Execution data queries cannot change the database session or role.",
      )
    }
    assertReadOnlySqlNode(child)
  }
}

/**
 * Identifies SQL-callable configuration paths that can change the active role.
 *
 * @param value - Parsed PostgreSQL function-call node.
 */
function isSessionMutationFunction(value: unknown) {
  if (!value || typeof value !== "object" || !("funcname" in value)) {
    return false
  }
  const funcname = (value as { funcname?: unknown }).funcname
  if (!Array.isArray(funcname)) return false
  const name = funcname
    .map((part) => {
      if (!part || typeof part !== "object" || !("String" in part)) return ""
      const stringNode = Reflect.get(part, "String")
      if (!stringNode || typeof stringNode !== "object") return ""
      const token = Reflect.get(stringNode, "sval")
      return typeof token === "string" ? token : ""
    })
    .filter(Boolean)
    .join(".")
    .toLowerCase()
  return name === "set_config" || name === "pg_catalog.set_config"
}

/**
 * Converts libpg_query node names to readable error text.
 *
 * @param type - Parser node name to humanize.
 */
function formatStatementType(type: string | undefined) {
  return (type ?? "unknown statement")
    .replace(/Stmt$/, "")
    .replaceAll(/([a-z])([A-Z])/g, "$1 $2")
    .toLowerCase()
}

/**
 * Converts PGlite results to the CLI's JSON-safe envelope.
 *
 * @param results - Raw results from one or more SQL statements.
 */
function normalizeQueryResults(results: Results[]) {
  return results.map((result) => ({
    affectedRows: result.affectedRows,
    fields: result.fields.map((field) => field.name),
    rows: result.rows.map((row) => normalizeJsonValue(row)),
  }))
}

/**
 * Recursively converts PostgreSQL values that JSON cannot serialize.
 *
 * @param value - Raw PGlite result value.
 */
function normalizeJsonValue(value: unknown): unknown {
  if (typeof value === "bigint") return value.toString()
  if (value instanceof Uint8Array) {
    return {
      $type: "bytes",
      base64: Buffer.from(value).toString("base64"),
      size: value.byteLength,
    }
  }
  if (value instanceof Date) return value.toISOString()
  if (Array.isArray(value)) return value.map(normalizeJsonValue)
  if (value && typeof value === "object") {
    return Object.fromEntries(
      Object.entries(value).map(([key, item]) => [
        key,
        normalizeJsonValue(item),
      ]),
    )
  }
  return value
}

/**
 * Converts a codec value to the dataset's loss-aware JSON projection.
 *
 * @param value - Decoded codec value.
 * @param encodedRef - Reference to the exact encoded owner column.
 */
export async function projectExecutionDataValue(
  value: Encodable,
  encodedRef: string,
): Promise<unknown> {
  if (value === undefined) return { $type: "undefined" }
  if (
    value === null ||
    typeof value === "boolean" ||
    typeof value === "string"
  ) {
    return value
  }
  if (typeof value === "bigint") {
    return { $type: "bigint", value: value.toString() }
  }
  if (typeof value === "number") {
    return Number.isFinite(value) && !Object.is(value, -0)
      ? value
      : {
          $type: "number",
          value: Object.is(value, -0) ? "-0" : String(value),
        }
  }
  if (value instanceof Date) {
    return { $type: "date", value: value.toISOString() }
  }
  if (value instanceof RegExp) {
    return { $type: "regexp", flags: value.flags, source: value.source }
  }
  if (value instanceof URL) return { $type: "url", value: value.href }
  if (value instanceof URLSearchParams) {
    return { $type: "url-search-params", entries: [...value.entries()] }
  }
  if (value instanceof Headers) {
    return { $type: "headers", entries: [...value.entries()] }
  }
  if (value instanceof Request) {
    return {
      $type: "request",
      body: projectBinary(
        new Uint8Array(await value.clone().arrayBuffer()),
        encodedRef,
      ),
      headers: Object.fromEntries(value.headers),
      method: value.method,
      url: value.url,
    }
  }
  if (value instanceof Response) {
    return {
      $type: "response",
      body: projectBinary(
        new Uint8Array(await value.clone().arrayBuffer()),
        encodedRef,
      ),
      headers: Object.fromEntries(value.headers),
      status: value.status,
      statusText: value.statusText,
    }
  }
  if (value instanceof File) {
    return {
      $type: "file",
      ...projectBinary(new Uint8Array(await value.arrayBuffer()), encodedRef),
      lastModified: value.lastModified,
      mediaType: value.type,
      name: value.name,
    }
  }
  if (value instanceof Blob) {
    return {
      $type: "blob",
      ...projectBinary(new Uint8Array(await value.arrayBuffer()), encodedRef),
      mediaType: value.type,
    }
  }
  if (value instanceof ArrayBuffer) {
    return {
      $type: "array-buffer",
      ...projectBinary(new Uint8Array(value), encodedRef),
    }
  }
  if (ArrayBuffer.isView(value)) {
    return {
      $type: value.constructor.name,
      ...projectBinary(
        new Uint8Array(value.buffer, value.byteOffset, value.byteLength),
        encodedRef,
      ),
    }
  }
  if (Array.isArray(value)) {
    return await Promise.all(
      value.map((item) => projectExecutionDataValue(item, encodedRef)),
    )
  }
  if (value instanceof Map) {
    return {
      $type: "map",
      entries: await Promise.all(
        [...value].map(async ([key, item]) => [
          await projectExecutionDataValue(key, encodedRef),
          await projectExecutionDataValue(item, encodedRef),
        ]),
      ),
    }
  }
  if (value instanceof Set) {
    return {
      $type: "set",
      values: await Promise.all(
        [...value].map((item) => projectExecutionDataValue(item, encodedRef)),
      ),
    }
  }

  const object = Object.fromEntries(
    await Promise.all(
      Object.entries(value).map(async ([key, item]) => [
        key,
        await projectExecutionDataValue(item, encodedRef),
      ]),
    ),
  )
  return Object.hasOwn(value, "$type")
    ? { $type: "object", value: object }
    : object
}

/**
 * Inlines small binary values and references the exact owner for large values.
 *
 * @param bytes - Decoded binary value.
 * @param encodedRef - Exact owner-column reference.
 */
function projectBinary(bytes: Uint8Array, encodedRef: string) {
  return bytes.byteLength <= INLINE_BINARY_LIMIT
    ? { base64: Buffer.from(bytes).toString("base64"), size: bytes.byteLength }
    : { encodedRef, omitted: true, size: bytes.byteLength }
}

/**
 * Decodes Electric's PostgreSQL hexadecimal bytea representation.
 *
 * @param value - Hexadecimal bytea token.
 * @throws When Electric uses an unsupported bytea representation.
 */
function parseBytea(value: string): Value<Uint8Array> {
  if (!value.startsWith("\\x")) {
    throw new Error("Electric returned an unsupported bytea representation.")
  }
  return new Uint8Array(Buffer.from(value.slice(2), "hex"))
}

/**
 * Requires a parsed Electric value to be binary.
 *
 * @param value - Parsed Electric column.
 * @param description - Column description for failures.
 * @throws When the value is not binary.
 */
function requireBytes(
  value: Value<Uint8Array> | undefined,
  description: string,
) {
  if (value instanceof Uint8Array) return value
  throw new Error(`Electric returned an invalid ${description}.`)
}

/**
 * Parses an optional binary Electric column.
 *
 * @param value - Parsed nullable Electric column.
 */
function optionalBytes(value: Value<Uint8Array> | undefined) {
  if (value === null || value === undefined) return null
  return requireBytes(value, "action output")
}

/**
 * Requires a parsed Electric value to be a string identifier.
 *
 * @param value - Parsed Electric column.
 * @param description - Identifier description for failures.
 * @throws When the value is not a string.
 */
function requireString(value: unknown, description: string) {
  if (typeof value === "string") return value
  throw new Error(`Electric returned an invalid ${description}.`)
}

/**
 * Serializes a derived JSON column while preserving SQL null.
 *
 * @param value - JSON-compatible projection.
 */
function stringifyJson(value: unknown) {
  return value === null ? null : JSON.stringify(value)
}

/** Returns the application-origin-specific data directory. */
function getOriginDirectory() {
  return path.join(
    EXECUTION_DATA_ROOT,
    createHash("sha256")
      .update(new URL(env.APP_ORIGIN).origin)
      .digest("base64url")
      .slice(0, 16),
  )
}

/**
 * Returns one authenticated profile's isolated directory.
 *
 * @param profileId - Opaque authorized profile identity.
 */
function getProfileDirectory(profileId: string) {
  return path.join(
    getOriginDirectory(),
    executionDatasetProfileIdSchema.parse(profileId),
  )
}

/**
 * Returns the PGlite data directory for one profile.
 *
 * @param profileId - Opaque authorized profile identity.
 */
function getDatabasePath(profileId: string) {
  return path.join(getProfileDirectory(profileId), "pgdata")
}

/** Returns the origin-local active-profile marker path. */
function getActiveProfilePath() {
  return path.join(getOriginDirectory(), "active-profile")
}

/**
 * Persists the active profile in owner-readable local metadata.
 *
 * @param profileId - Opaque authorized profile identity.
 */
async function rememberProfile(profileId: string) {
  const directory = getOriginDirectory()
  await fs.mkdir(directory, { mode: 0o700, recursive: true })
  await fs.chmod(directory, 0o700)
  await atomicWritePrivateFile(getActiveProfilePath(), profileId)
  saveExecutionDataProfileSync(profileId)
}

/** Reads and validates the origin-local active-profile marker. */
async function readActiveProfile() {
  try {
    const result = executionDatasetProfileIdSchema.safeParse(
      (await fs.readFile(getActiveProfilePath(), "utf8")).trim(),
    )
    return result.success ? result.data : null
  } catch {
    return null
  }
}

/**
 * Removes one derived local profile and its matching active marker.
 *
 * @param profileId - Opaque profile identity to remove.
 */
async function removeExecutionDataProfile(profileId: string) {
  const [activeProfile] = await Promise.all([
    readActiveProfile(),
    fs.rm(getProfileDirectory(profileId), { force: true, recursive: true }),
  ])
  if (activeProfile === profileId) {
    await fs.rm(getActiveProfilePath(), { force: true })
  }
}

/**
 * Atomically replaces a sensitive file with owner-only permissions.
 *
 * @param filePath - Destination file to replace.
 * @param contents - Complete string or binary contents.
 */
export async function atomicWritePrivateFile(
  filePath: string,
  contents: string | Uint8Array,
) {
  const temporaryPath = path.join(
    path.dirname(filePath),
    `.${path.basename(filePath)}.${randomUUID()}.tmp`,
  )
  try {
    await fs.writeFile(temporaryPath, contents, { mode: 0o600 })
    await fs.chmod(temporaryPath, 0o600)
    await fs.rename(temporaryPath, filePath)
  } catch (error) {
    await fs.rm(temporaryPath, { force: true })
    throw error
  }
}