import { DatabaseSync } from 'node:sqlite'
import { chmodSync, existsSync, mkdirSync, statSync } from 'node:fs'
import { join } from 'node:path'
import electron from 'electron'
import type {
  CloudProjectionCheckpoint,
  CloudProjectionFenceToken,
  WorkflowDetail,
  WorkflowDraft,
  WorkflowListPayload,
  WorkflowRun,
  WorkflowRunStatus,
  WorkflowStatus,
  WorkflowSummary,
  WorkflowTrigger,
  WorkflowTriggerType,
} from '@open-cowork/shared'
import {
  createCloudProjectionCheckpoint,
  createCloudProjectionFenceToken,
} from '@open-cowork/shared'
import { getAppDataDir } from '../config-loader.ts'
import {
  readSafeStorageBackendForPolicy,
  resolveSecretStorageMode,
} from '../secure-storage-policy.ts'
import { computeNextWorkflowRunAt } from './workflow-schedule.ts'
import {
  assertWorkflowCapabilities,
  isWorkflowTriggerType,
  MAX_WORKFLOW_LIST_ITEMS,
  normalizeWorkflowDraft,
  previewWorkflowDraft as previewWorkflowDraftCalculation,
  type WorkflowDraftNormalizationOptions,
} from './workflow-normalization.ts'
import {
  parseWorkflowTriggersFromStorageWithAdapter,
  serializeWorkflowTriggersForStorageWithAdapter,
  type WorkflowSecretStorageAdapter,
} from './workflow-secret-storage.ts'

const WORKFLOW_DB_SCHEMA_VERSION = 1
const WORKFLOW_SCHEMA_VERSION_KEY = 'schema_version'
const WORKFLOW_PROJECTION_VERSION_KEY = 'workflow_projection_version'
const LOCAL_WORKFLOW_PROJECTION_TENANT_ID = 'desktop-local'
const VALID_STATUS = new Set<WorkflowStatus>(['active', 'paused', 'running', 'failed', 'archived'])
const VALID_RUN_STATUS = new Set<WorkflowRunStatus>(['queued', 'running', 'completed', 'failed', 'cancelled'])

let workflowDb: DatabaseSync | null = null
let workflowDbForTests: DatabaseSync | null = null
let transactionCounter = 0
let workflowSecretStorageForTests: WorkflowSecretStorageAdapter | null = null

const electronApp = (electron as { app?: typeof import('electron').app }).app
const electronSafeStorage = (electron as { safeStorage?: typeof import('electron').safeStorage }).safeStorage
const electronSafeStorageBackend = electronSafeStorage as (typeof import('electron').safeStorage & {
  getSelectedStorageBackend?: () => string
}) | undefined

type DbRow = Record<string, unknown>
type WorkflowWriteOptions = WorkflowDraftNormalizationOptions
type WorkflowDraftOptions = WorkflowDraftNormalizationOptions
export type { WorkflowCapabilityValidationContext } from './workflow-normalization.ts'

function workflowDbPath() {
  const dir = getAppDataDir()
  mkdirSync(dir, { recursive: true })
  return join(dir, 'workflows.sqlite')
}

function ensureWorkflowDbFileModes(dbPath = workflowDbPath()) {
  if (process.platform === 'win32') return
  for (const path of [dbPath, `${dbPath}-wal`, `${dbPath}-shm`]) {
    if (!existsSync(path)) continue
    chmodSync(path, 0o600)
  }
}

function parseJson<T>(value: unknown, fallback: T): T {
  if (typeof value !== 'string') return fallback
  try {
    return JSON.parse(value) as T
  } catch {
    return fallback
  }
}

function getWorkflowSecretStorage(): WorkflowSecretStorageAdapter {
  if (workflowSecretStorageForTests) return workflowSecretStorageForTests
  const mode = resolveSecretStorageMode({
    isPackaged: Boolean(electronApp?.isPackaged),
    encryptionAvailable: Boolean(electronSafeStorage?.isEncryptionAvailable?.()),
    selectedStorageBackend: readSafeStorageBackendForPolicy(
      electronSafeStorageBackend?.getSelectedStorageBackend?.bind(electronSafeStorageBackend),
    ),
  })
  return {
    mode,
    encryptString: electronSafeStorage?.encryptString?.bind(electronSafeStorage),
    decryptString: electronSafeStorage?.decryptString?.bind(electronSafeStorage),
  }
}

export function serializeWorkflowTriggersForStorage(triggers: WorkflowTrigger[]) {
  return serializeWorkflowTriggersForStorageWithAdapter(triggers, getWorkflowSecretStorage())
}

export function parseWorkflowTriggersFromStorage(value: unknown) {
  return parseWorkflowTriggersFromStorageWithAdapter(value, getWorkflowSecretStorage())
}

export function setWorkflowSecretStorageForTests(adapter: WorkflowSecretStorageAdapter | null) {
  workflowSecretStorageForTests = adapter
}

export function setWorkflowDatabaseForTests(db: DatabaseSync | null) {
  workflowDb?.close()
  workflowDb = null
  workflowDbForTests = db
  transactionCounter = 0
  if (db) initDb(db)
}

function writeNow(options?: WorkflowWriteOptions) {
  return options?.now ?? new Date()
}

function randomWebhookSecret() {
  return crypto.randomUUID().replace(/-/g, '') + crypto.randomUUID().replace(/-/g, '')
}

function projectDirectoryExists(directory: string) {
  try {
    return statSync(directory).isDirectory()
  } catch {
    return false
  }
}

function workflowDraftOptions(options?: WorkflowDraftOptions): WorkflowDraftNormalizationOptions {
  return {
    ...options,
    projectDirectoryExists: options?.projectDirectoryExists ?? projectDirectoryExists,
  }
}

function initDb(db: DatabaseSync) {
  db.exec(`
    create table if not exists workflow_meta (
      key text primary key,
      value text not null
    );
    create table if not exists workflows (
      id text primary key,
      title text not null,
      instructions text not null,
      agent_name text not null,
      skill_names_json text not null,
      tool_ids_json text not null,
      status text not null,
      project_directory text,
      draft_session_id text,
      triggers_json text not null,
      created_at text not null,
      updated_at text not null,
      next_run_at text,
      last_run_at text,
      latest_run_id text,
      latest_run_status text,
      latest_run_session_id text,
      latest_run_summary text
    );
    create table if not exists workflow_runs (
      id text primary key,
      workflow_id text not null,
      session_id text,
      trigger_type text not null,
      trigger_payload_json text,
      status text not null,
      title text not null,
      summary text,
      error text,
      created_at text not null,
      started_at text,
      finished_at text
    );
    create index if not exists idx_workflow_runs_workflow on workflow_runs(workflow_id, created_at);
    create index if not exists idx_workflows_due on workflows(status, next_run_at);
  `)
  db.prepare(`
    insert into workflow_meta (key, value)
    values (?, ?)
    on conflict(key) do update set value = excluded.value
  `).run(WORKFLOW_SCHEMA_VERSION_KEY, String(WORKFLOW_DB_SCHEMA_VERSION))
}

export function getWorkflowDb() {
  if (workflowDbForTests) return workflowDbForTests
  if (workflowDb) return workflowDb
  const dbPath = workflowDbPath()
  const db = new DatabaseSync(dbPath)
  try {
    db.exec('pragma journal_mode = WAL;')
    initDb(db)
    ensureWorkflowDbFileModes(dbPath)
    workflowDb = db
    return db
  } catch (error) {
    db.close()
    throw error
  }
}

function withTransaction<T>(callback: (db: DatabaseSync) => T): T {
  const db = getWorkflowDb()
  const savepoint = `workflow_tx_${transactionCounter += 1}`
  db.exec(`savepoint ${savepoint}`)
  try {
    const result = callback(db)
    db.exec(`release savepoint ${savepoint}`)
    if (!workflowDbForTests) ensureWorkflowDbFileModes()
    return result
  } catch (error) {
    try {
      db.exec(`rollback to savepoint ${savepoint}`)
    } finally {
      db.exec(`release savepoint ${savepoint}`)
      if (!workflowDbForTests) ensureWorkflowDbFileModes()
    }
    throw error
  }
}

function readWorkflowProjectionVersion(db = getWorkflowDb()) {
  const row = db.prepare('select value from workflow_meta where key = ?').get(WORKFLOW_PROJECTION_VERSION_KEY) as { value?: unknown } | undefined
  const version = Number.parseInt(String(row?.value || '0'), 10)
  return Number.isFinite(version) && version >= 0 ? version : 0
}

function bumpWorkflowProjectionVersion(db: DatabaseSync) {
  const next = readWorkflowProjectionVersion(db) + 1
  db.prepare(`
    insert into workflow_meta (key, value)
    values (?, ?)
    on conflict(key) do update set value = excluded.value
  `).run(WORKFLOW_PROJECTION_VERSION_KEY, String(next))
  return next
}

function workflowRunProjectionFence(run: Pick<WorkflowRun, 'id' | 'workflowId'>, projectionVersion = readWorkflowProjectionVersion()): CloudProjectionFenceToken {
  return createCloudProjectionFenceToken({
    scope: 'workflow-run',
    tenantId: LOCAL_WORKFLOW_PROJECTION_TENANT_ID,
    workflowId: run.workflowId,
    runId: run.id,
    sequence: projectionVersion,
    projectionVersion,
    checkpointVersion: projectionVersion,
  })
}

function withWorkflowRunProjectionFence<T extends WorkflowRun | null>(run: T, projectionVersion?: number): T {
  if (!run) return run
  return {
    ...run,
    projectionFence: workflowRunProjectionFence(run, projectionVersion),
  }
}

export function getWorkflowRunProjectionCheckpoint(runId: string): CloudProjectionCheckpoint | null {
  const run = getWorkflowRun(runId)
  if (!run) return null
  const projectionVersion = readWorkflowProjectionVersion()
  return createCloudProjectionCheckpoint({
    scope: 'workflow-run',
    tenantId: LOCAL_WORKFLOW_PROJECTION_TENANT_ID,
    workflowId: run.workflowId,
    runId: run.id,
    sequence: projectionVersion,
    projectionVersion,
    checkpointVersion: projectionVersion,
    updatedAt: run.finishedAt || run.startedAt || run.createdAt || new Date().toISOString(),
  })
}

function webhookUrlForWorkflow(workflow: WorkflowSummary, webhookBaseUrl?: string | null) {
  if (!webhookBaseUrl) return null
  const webhook = workflow.triggers.find((trigger) => (
    trigger.enabled
    && trigger.type === 'webhook'
    && typeof trigger.webhookSecret === 'string'
    && trigger.webhookSecret.length > 0
  ))
  return webhook ? `${webhookBaseUrl}/workflows/${encodeURIComponent(workflow.id)}` : null
}

function rowToWorkflow(row: DbRow, webhookBaseUrl?: string | null): WorkflowSummary {
  const status = VALID_STATUS.has(String(row.status) as WorkflowStatus) ? String(row.status) as WorkflowStatus : 'active'
  const latestRunStatus = row.latest_run_status && VALID_RUN_STATUS.has(String(row.latest_run_status) as WorkflowRunStatus)
    ? String(row.latest_run_status) as WorkflowRunStatus
    : null
  const workflow: WorkflowSummary = {
    id: String(row.id || ''),
    title: String(row.title || ''),
    instructions: String(row.instructions || ''),
    agentName: String(row.agent_name || 'build'),
    skillNames: parseJson<string[]>(row.skill_names_json, []),
    toolIds: parseJson<string[]>(row.tool_ids_json, []),
    status,
    projectDirectory: typeof row.project_directory === 'string' ? row.project_directory : null,
    draftSessionId: typeof row.draft_session_id === 'string' ? row.draft_session_id : null,
    triggers: parseWorkflowTriggersFromStorage(row.triggers_json),
    createdAt: String(row.created_at || ''),
    updatedAt: String(row.updated_at || ''),
    nextRunAt: typeof row.next_run_at === 'string' ? row.next_run_at : null,
    lastRunAt: typeof row.last_run_at === 'string' ? row.last_run_at : null,
    latestRunId: typeof row.latest_run_id === 'string' ? row.latest_run_id : null,
    latestRunStatus,
    latestRunSessionId: typeof row.latest_run_session_id === 'string' ? row.latest_run_session_id : null,
    latestRunSummary: typeof row.latest_run_summary === 'string' ? row.latest_run_summary : null,
    webhookUrl: null,
  }
  workflow.webhookUrl = webhookUrlForWorkflow(workflow, webhookBaseUrl)
  return workflow
}

function rowToRun(row: DbRow): WorkflowRun {
  const status = VALID_RUN_STATUS.has(String(row.status) as WorkflowRunStatus) ? String(row.status) as WorkflowRunStatus : 'queued'
  const rawTriggerType = String(row.trigger_type)
  const triggerType = isWorkflowTriggerType(rawTriggerType)
    ? rawTriggerType
    : 'manual'
  return {
    id: String(row.id || ''),
    workflowId: String(row.workflow_id || ''),
    sessionId: typeof row.session_id === 'string' ? row.session_id : null,
    triggerType,
    triggerPayload: parseJson<Record<string, unknown> | null>(row.trigger_payload_json, null),
    status,
    title: String(row.title || ''),
    summary: typeof row.summary === 'string' ? row.summary : null,
    error: typeof row.error === 'string' ? row.error : null,
    createdAt: String(row.created_at || ''),
    startedAt: typeof row.started_at === 'string' ? row.started_at : null,
    finishedAt: typeof row.finished_at === 'string' ? row.finished_at : null,
  }
}

function listRunsForWorkflow(workflowId: string, limit = 25) {
  const rows = getWorkflowDb().prepare('select * from workflow_runs where workflow_id = ? order by created_at desc limit ?')
    .all(workflowId, limit) as DbRow[]
  return rows.map(rowToRun)
}

export function previewWorkflowDraft(draft: WorkflowDraft, options?: WorkflowDraftOptions) {
  return previewWorkflowDraftCalculation(draft, workflowDraftOptions(options))
}

export function listWorkflows(webhookBaseUrl?: string | null): WorkflowListPayload {
  const workflows = (getWorkflowDb().prepare('select * from workflows order by updated_at desc').all() as DbRow[])
    .map((row) => rowToWorkflow(row, webhookBaseUrl))
  const runs = (getWorkflowDb().prepare('select * from workflow_runs order by created_at desc limit 100').all() as DbRow[])
    .map(rowToRun)
  return { workflows, runs }
}

export function getWorkflow(workflowId: string, webhookBaseUrl?: string | null): WorkflowDetail | null {
  const row = getWorkflowDb().prepare('select * from workflows where id = ?').get(workflowId) as DbRow | undefined
  if (!row) return null
  return {
    ...rowToWorkflow(row, webhookBaseUrl),
    runs: listRunsForWorkflow(workflowId),
  }
}

export function createWorkflow(draft: WorkflowDraft, webhookBaseUrl?: string | null, options?: WorkflowWriteOptions): WorkflowDetail {
  const nowDate = writeNow(options)
  const normalized = normalizeWorkflowDraft(draft, workflowDraftOptions({ ...options, now: nowDate }))
  assertWorkflowCapabilities(normalized, workflowDraftOptions(options))
  const now = nowDate.toISOString()
  const id = crypto.randomUUID()
  const nextRunAt = computeNextWorkflowRunAt(normalized.triggers, nowDate)
  withTransaction((db) => {
    db.prepare(`
      insert into workflows (
        id, title, instructions, agent_name, skill_names_json, tool_ids_json, status,
        project_directory, draft_session_id, triggers_json, created_at, updated_at,
        next_run_at, last_run_at, latest_run_id, latest_run_status, latest_run_session_id, latest_run_summary
      ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, null, null, null, null, null)
    `).run(
      id,
      normalized.title,
      normalized.instructions,
      normalized.agentName,
      JSON.stringify(normalized.skillNames || []),
      JSON.stringify(normalized.toolIds || []),
      'active',
      normalized.projectDirectory || null,
      normalized.draftSessionId || null,
      serializeWorkflowTriggersForStorage(normalized.triggers),
      now,
      now,
      nextRunAt,
    )
    bumpWorkflowProjectionVersion(db)
  })
  return getWorkflow(id, webhookBaseUrl)!
}

export function updateWorkflowStatus(workflowId: string, status: WorkflowStatus, webhookBaseUrl?: string | null, options?: WorkflowWriteOptions) {
  if (!VALID_STATUS.has(status)) throw new Error('Workflow status is invalid.')
  const nowDate = writeNow(options)
  const now = nowDate.toISOString()
  withTransaction((db) => {
    const row = db.prepare('select triggers_json from workflows where id = ?').get(workflowId) as DbRow | undefined
    const triggers = parseWorkflowTriggersFromStorage(row?.triggers_json)
    const nextRunAt = status === 'active' ? computeNextWorkflowRunAt(triggers, nowDate) : null
    db.prepare('update workflows set status = ?, updated_at = ?, next_run_at = ? where id = ?')
      .run(status, now, nextRunAt, workflowId)
    bumpWorkflowProjectionVersion(db)
  })
  return getWorkflow(workflowId, webhookBaseUrl)
}

export function regenerateWorkflowWebhookSecret(workflowId: string, webhookBaseUrl?: string | null) {
  const detail = getWorkflow(workflowId, webhookBaseUrl)
  if (!detail) return null
  const triggers = detail.triggers.map((trigger) => trigger.type === 'webhook'
    ? { ...trigger, webhookSecret: randomWebhookSecret() }
    : trigger)
  const now = new Date().toISOString()
  withTransaction((db) => {
    db.prepare('update workflows set triggers_json = ?, updated_at = ? where id = ?')
      .run(serializeWorkflowTriggersForStorage(triggers), now, workflowId)
    bumpWorkflowProjectionVersion(db)
  })
  return getWorkflow(workflowId, webhookBaseUrl)
}

export function listDueWorkflows(now = new Date(), webhookBaseUrl?: string | null) {
  const rows = getWorkflowDb().prepare(`
    select * from workflows
    where status = 'active'
      and next_run_at is not null
      and next_run_at <= ?
    order by next_run_at asc
  `).all(now.toISOString()) as DbRow[]
  return rows.map((row) => rowToWorkflow(row, webhookBaseUrl))
}

export function createWorkflowRun(workflowId: string, triggerType: WorkflowTriggerType, payload: Record<string, unknown> | null = null) {
  const workflow = getWorkflow(workflowId)
  if (!workflow) throw new Error('Workflow not found.')
  if (workflow.status === 'archived') throw new Error('Archived workflows cannot run.')
  if (workflow.status === 'paused') throw new Error('Paused workflows cannot run.')
  if (workflow.status === 'running') throw new Error('Workflow is already running.')
  const now = new Date().toISOString()
  const id = crypto.randomUUID()
  let projectionVersion = 0
  withTransaction((db) => {
    db.prepare(`
      insert into workflow_runs (
        id, workflow_id, session_id, trigger_type, trigger_payload_json, status, title,
        summary, error, created_at, started_at, finished_at
      ) values (?, ?, null, ?, ?, 'queued', ?, null, null, ?, null, null)
    `).run(id, workflowId, triggerType, payload ? JSON.stringify(payload) : null, `Run ${workflow.title}`, now)
    db.prepare(`
      update workflows
      set status = 'running', latest_run_id = ?, latest_run_status = 'queued', updated_at = ?
      where id = ?
    `).run(id, now, workflowId)
    projectionVersion = bumpWorkflowProjectionVersion(db)
  })
  return withWorkflowRunProjectionFence(getWorkflowRun(id), projectionVersion)
}

export function claimDueWorkflowRun(now = new Date()) {
  const claimedAt = now.toISOString()
  return withTransaction((db) => {
    for (let attempt = 0; attempt < MAX_WORKFLOW_LIST_ITEMS; attempt += 1) {
      const row = db.prepare(`
        select * from workflows
        where status = 'active'
          and next_run_at is not null
          and next_run_at <= ?
        order by next_run_at asc
        limit 1
      `).get(claimedAt) as DbRow | undefined
      if (!row) return null

      const workflow = rowToWorkflow(row)
      const trigger = workflow.triggers.find((entry) => (
        entry.enabled && entry.type === 'schedule' && entry.schedule && workflow.nextRunAt
      ))
      if (!trigger || !workflow.nextRunAt) {
        db.prepare('update workflows set next_run_at = ?, updated_at = ? where id = ?')
          .run(computeNextWorkflowRunAt(workflow.triggers, now), claimedAt, workflow.id)
        continue
      }

      const runId = crypto.randomUUID()
      const update = db.prepare(`
        update workflows
        set status = 'running', latest_run_id = ?, latest_run_status = 'queued', updated_at = ?
        where id = ?
          and status = 'active'
          and next_run_at = ?
      `).run(runId, claimedAt, workflow.id, workflow.nextRunAt)
      if (Number(update.changes) !== 1) continue

      const payload = {
        source: 'schedule',
        scheduledFor: workflow.nextRunAt,
      }
      db.prepare(`
        insert into workflow_runs (
          id, workflow_id, session_id, trigger_type, trigger_payload_json, status, title,
          summary, error, created_at, started_at, finished_at
        ) values (?, ?, null, 'schedule', ?, 'queued', ?, null, null, ?, null, null)
      `).run(runId, workflow.id, JSON.stringify(payload), `Run ${workflow.title}`, claimedAt)

      const projectionVersion = bumpWorkflowProjectionVersion(db)
      return withWorkflowRunProjectionFence(getWorkflowRun(runId), projectionVersion)
    }
    return null
  })
}

export function getWorkflowRun(runId: string) {
  const row = getWorkflowDb().prepare('select * from workflow_runs where id = ?').get(runId) as DbRow | undefined
  return row ? rowToRun(row) : null
}

export function attachWorkflowRunSession(workflowId: string, runId: string, sessionId: string) {
  const now = new Date().toISOString()
  let projectionVersion = 0
  withTransaction((db) => {
    db.prepare('update workflow_runs set session_id = ?, status = ?, started_at = coalesce(started_at, ?) where id = ?')
      .run(sessionId, 'running', now, runId)
    db.prepare(`
      update workflows
      set latest_run_id = ?, latest_run_status = ?, latest_run_session_id = ?, updated_at = ?, status = ?
      where id = ?
    `).run(runId, 'running', sessionId, now, 'running', workflowId)
    projectionVersion = bumpWorkflowProjectionVersion(db)
  })
  return withWorkflowRunProjectionFence(getWorkflowRun(runId), projectionVersion)
}

export function markWorkflowRunCompleted(runId: string, summary: string | null) {
  const run = getWorkflowRun(runId)
  if (!run) return null
  if (run.status === 'completed' || run.status === 'failed' || run.status === 'cancelled') return run
  const now = new Date().toISOString()
  const workflow = getWorkflow(run.workflowId)
  const nextStatus = workflow?.status === 'paused' || workflow?.status === 'archived'
    ? workflow.status
    : 'active'
  const nextRunAt = nextStatus === 'active' && workflow
    ? computeNextWorkflowRunAt(workflow.triggers, new Date(now))
    : null
  let projectionVersion = 0
  withTransaction((db) => {
    db.prepare('update workflow_runs set status = ?, summary = ?, finished_at = ? where id = ?')
      .run('completed', summary, now, runId)
    db.prepare(`
      update workflows
      set status = ?, latest_run_id = ?, latest_run_status = 'completed',
        latest_run_summary = ?, last_run_at = ?, next_run_at = ?, updated_at = ?
      where id = ?
    `).run(nextStatus, runId, summary, now, nextRunAt, now, run.workflowId)
    projectionVersion = bumpWorkflowProjectionVersion(db)
  })
  return withWorkflowRunProjectionFence(getWorkflowRun(runId), projectionVersion)
}

export function markWorkflowRunFailed(runId: string, error: string) {
  const run = getWorkflowRun(runId)
  if (!run) return null
  if (run.status === 'completed' || run.status === 'failed' || run.status === 'cancelled') return run
  const now = new Date().toISOString()
  const workflow = getWorkflow(run.workflowId)
  const nextRunAt = workflow ? computeNextWorkflowRunAt(workflow.triggers, new Date(now)) : null
  const nextStatus = workflow?.status === 'paused' || workflow?.status === 'archived'
    ? workflow.status
    : 'active'
  let projectionVersion = 0
  withTransaction((db) => {
    db.prepare('update workflow_runs set status = ?, error = ?, finished_at = ? where id = ?')
      .run('failed', error, now, runId)
    db.prepare(`
      update workflows
      set status = ?, latest_run_id = ?, latest_run_status = 'failed',
        latest_run_summary = ?, next_run_at = ?, updated_at = ?
      where id = ?
    `).run(nextStatus, runId, error, nextRunAt, now, run.workflowId)
    projectionVersion = bumpWorkflowProjectionVersion(db)
  })
  return withWorkflowRunProjectionFence(getWorkflowRun(runId), projectionVersion)
}

export function recoverInterruptedWorkflowRuns(error = 'Workflow run was interrupted before completion.', now = new Date()) {
  const finishedAt = now.toISOString()
  const rows = getWorkflowDb().prepare(`
    select * from workflow_runs
    where status in ('queued', 'running')
    order by created_at asc
  `).all() as DbRow[]
  if (rows.length === 0) return []

  return withTransaction((db) => {
    const recovered: WorkflowRun[] = []
    for (const row of rows) {
      const run = rowToRun(row)
      db.prepare('update workflow_runs set status = ?, error = ?, finished_at = ? where id = ?')
        .run('failed', error, finishedAt, run.id)

      const workflowRow = db.prepare('select * from workflows where id = ?').get(run.workflowId) as DbRow | undefined
      if (workflowRow) {
        const status = VALID_STATUS.has(String(workflowRow.status) as WorkflowStatus)
          ? String(workflowRow.status) as WorkflowStatus
          : 'active'
        const nextStatus = status === 'paused' || status === 'archived' ? status : 'active'
        const triggers = parseWorkflowTriggersFromStorage(workflowRow.triggers_json)
        const nextRunAt = nextStatus === 'active' ? computeNextWorkflowRunAt(triggers, now) : null
        db.prepare(`
          update workflows
          set status = ?, latest_run_id = ?, latest_run_status = 'failed',
            latest_run_summary = ?, next_run_at = ?, updated_at = ?
          where id = ? and latest_run_id = ?
        `).run(nextStatus, run.id, error, nextRunAt, finishedAt, run.workflowId, run.id)
      }

      const projectionVersion = bumpWorkflowProjectionVersion(db)
      recovered.push(withWorkflowRunProjectionFence(rowToRun({
        ...row,
        status: 'failed',
        error,
        finished_at: finishedAt,
      }), projectionVersion))
    }
    return recovered
  })
}

export function clearWorkflowStoreCache() {
  workflowDb?.close()
  workflowDb = null
  workflowDbForTests = null
  transactionCounter = 0
}
