// ---------------------------------------------------------------------------
// Data migration: rebuild tool catalogs whose operations return NDJSON.
//
// Extraction used to persist an NDJSON operation's PER-LINE response schema as
// the tool's output schema, while the invoke path returns an ARRAY of parsed
// lines, so describe previews promised a single object that invocations never
// returned. The producer now wraps those schemas in an array; the per-line
// shape can't be recognized from a persisted `tool` row alone (it's just a
// JSON schema), but the stored operation bindings kept the response content
// type, so the affected connections are findable and the catalog machinery
// already knows how to rebuild them: stale-mark `tools_synced_at` and the next
// read re-produces the rows through the fixed producer. Mirrors the cloud
// drizzle migration (apps/cloud/drizzle/0010_ndjson_output_arrays.sql).
//
// Effectively idempotent: re-running re-marks the same connections, and a
// rebuild from unchanged bindings writes identical rows.
// ---------------------------------------------------------------------------

import { Effect } from "effect";
import { DataMigrationError, type SqliteDataMigrationClient } from "@executor-js/sdk/core";

const MIGRATION_NAME = "2026-07-09-openapi-ndjson-output-arrays";

// Substring probes for the NDJSON media types (NDJSON_MEDIA_TYPES in
// openapi-utils.ts). The binding JSON inside `plugin_storage.data` may be a
// nested object or a re-encoded string, so a LIKE over the raw row text is the
// shape-agnostic match; false positives only cause a harmless extra rebuild.
const NDJSON_TYPE_PROBES = [
  "application/stream+json",
  "application/x-ndjson",
  "application/jsonl",
] as const;

const execute = (
  client: SqliteDataMigrationClient,
  stmt: string | { readonly sql: string; readonly args: readonly unknown[] },
) =>
  Effect.tryPromise({
    try: () => client.execute(stmt),
    catch: (cause) => new DataMigrationError({ migration: MIGRATION_NAME, cause }),
  });

const tableExists = (client: SqliteDataMigrationClient, table: string) =>
  Effect.map(
    execute(client, {
      sql: "SELECT name FROM sqlite_master WHERE type = 'table' AND name = ?",
      args: [table],
    }),
    (result) => result.rows.length > 0,
  );

/** Stale-mark every connection whose integration has at least one stored
 *  operation binding with an NDJSON response, so the next read rebuilds its
 *  tool rows (with array-wrapped output schemas). Returns the number of
 *  connections marked. Fresh databases may lack either table; nothing to
 *  migrate. */
export const runSqliteNdjsonOutputMigration = (
  client: SqliteDataMigrationClient,
): Effect.Effect<number, DataMigrationError> =>
  Effect.gen(function* () {
    for (const table of ["connection", "plugin_storage"]) {
      if (!(yield* tableExists(client, table))) return 0;
    }

    const probes = NDJSON_TYPE_PROBES.map(() => "ps.data LIKE ?").join(" OR ");
    const affected = yield* execute(client, {
      sql: `SELECT DISTINCT ps.tenant AS tenant, json_extract(ps.data, '$.integration') AS integration
            FROM plugin_storage ps
            WHERE ps.collection = 'operation' AND (${probes})`,
      args: NDJSON_TYPE_PROBES.map((probe) => `%${probe}%`),
    });

    let marked = 0;
    for (const row of affected.rows) {
      if (typeof row.tenant !== "string" || typeof row.integration !== "string") continue;
      yield* execute(client, {
        sql: "UPDATE connection SET tools_synced_at = NULL WHERE tenant = ? AND integration = ?",
        args: [row.tenant, row.integration],
      });
      marked += 1;
    }
    return marked;
  });

/** Registry entry for the boot-time data-migration ledger. */
export const openApiNdjsonOutputDataMigration = {
  name: MIGRATION_NAME,
  run: (client: SqliteDataMigrationClient) =>
    runSqliteNdjsonOutputMigration(client).pipe(Effect.asVoid),
};