var validation = require("./knowledge-db-validation");
var input = require("./knowledge-db-import-validation");

// importRecords({ databaseId, expectedDatabaseRevision, operationId, sourceNamespace,
// rows: [{ sourceKey, sourcePayload, values }], actor }) creates only unseen source rows.
// Both sourcePayload and supplied values are immutable import identity; corrections use CRUD.
// Record references can use sourceNamespace/sourceKey instead of id, including forward
// references within this batch. Cross-database sources must already be imported locally.
// readImportProgress({ databaseId, sourceNamespace, offset?, limit? }) returns native ids.
// Import mappings survive receipt expiry; replay with a fresh operation id and current revision.
// Limits: 100 rows / 256 KiB of row JSON per batch; shared 20000-record / 16 MiB
// aggregate bounds include archives, history, mappings and receipts (500 live / 24 hours).
// Each fresh operation retains one receipt and uses one canonical replacement; source
// replay preserves the revision. New rows also use the existing optional summary
// projection write. No filesystem input is accepted.
function attachKnowledgeDbImport(ctx) {
  function mappings(database) { return database.importMappings || []; }
  function index(database) {
    var out = new Map(); mappings(database).forEach(function (mapping) { out.set(input.identity(mapping.sourceNamespace, mapping.sourceKey), mapping); }); return out;
  }
  function convert(values, schema, databases, next, planned) {
    var fields = {}; schema.forEach(function (field) { fields[field.id] = field; });
    var out = Object.assign({}, values);
    Object.keys(out).forEach(function (fieldId) {
      var field = fields[fieldId]; var value = out[fieldId];
      if (!field || field.type !== "reference" || !value || !Object.prototype.hasOwnProperty.call(value, "sourceKey")) return;
      validation.plainObject(value, "Import reference");
      if (value.type !== "record" || value.id !== undefined) throw ctx.error("Source references require type record and no native id.", "INVALID");
      var databaseId = validation.id(value.databaseId || field.targetDatabaseId || next._id, "reference database id");
      var namespace = input.sourceText(value.sourceNamespace, "Reference source namespace", 120);
      var key = input.sourceText(value.sourceKey, "Reference source key", 200);
      var target = databases.find(function (database) { return database._id === databaseId; });
      var mapping = databaseId === next._id ? planned.get(input.identity(namespace, key)) : target && index(target).get(input.identity(namespace, key));
      if (!mapping) throw ctx.error("Reference source row is outside this Mate or has not been imported.", "FORBIDDEN_REFERENCE");
      out[fieldId] = { type: "record", databaseId: databaseId, id: mapping.recordId };
    });
    return out;
  }
  function importRecords(args) {
    // Snapshot caller-owned JSON before any asynchronous operation.
    var revalidate = args.revalidate; var batch = ctx.clone(input.batch(args)); args = ctx.clone(args); args.revalidate = revalidate;
    var operationId = ctx.mutationId(args.operationId); var fingerprint = ctx.fingerprint("import", args);
    return ctx.serialize(args.databaseId, function () { return ctx.load(args.databaseId).then(async function (current) {
      ctx.guard(args, current);
      var duplicate = ctx.receipt(current, operationId, fingerprint);
      if (duplicate) return Object.assign({}, duplicate, { projectionPending: current.projectionPending === true, projectionError: current.projectionError || null });
      if (current.archivedAt) throw ctx.error("Knowledge database is archived.", "ARCHIVED", current.revision);
      ctx.ensureRevision(args.expectedDatabaseRevision, current.revision, "database");
      var next = ctx.clone(current); var planned = index(current); var added = [];
      var resultMappings = batch.rows.map(function (row) {
        var identity = input.identity(batch.sourceNamespace, row.sourceKey); var existing = planned.get(identity);
        if (existing && existing.fingerprint !== row.fingerprint) throw ctx.error("Source row was already imported with a different payload: " + row.sourceKey, "IMPORT_SOURCE_CONFLICT");
        if (existing) return Object.assign({}, existing, { created: false });
        var mapping = { sourceNamespace: batch.sourceNamespace, sourceKey: row.sourceKey, fingerprint: row.fingerprint, recordId: ctx.newId("rec") };
        planned.set(identity, mapping); added.push({ row: row, mapping: mapping });
        return Object.assign({}, mapping, { created: true });
      });
      if (next.records.length + added.length > validation.MAX_RECORDS) throw ctx.error("Database exceeds the " + validation.MAX_RECORDS + " record limit (including archived records).", "LIMIT");
      var result = { databaseId: current._id, databaseRevision: current.revision, sourceNamespace: batch.sourceNamespace, createdCount: added.length, replayedCount: batch.rows.length - added.length, mappings: resultMappings };
      // Remember fresh replay operations too, so reusing an operation id with another
      // payload still conflicts. Preserve record values, archives and database revision.
      if (!added.length) {
        ctx.remember(next, operationId, fingerprint, result);
        await ctx.replace(current, next, args);
        return Object.assign(result, { projectionPending: current.projectionPending === true, projectionError: current.projectionError || null });
      }
      var databases = await ctx.all(); ctx.guard(args, current); var at = Date.now(); var who = ctx.actor(args.actor);
      added.forEach(function (item) {
        next.records.push({ id: item.mapping.recordId, revision: 1, values: {}, archivedAt: null, createdAt: at, updatedAt: at, createdBy: who, updatedBy: who });
        next.recordHistory[item.mapping.recordId] = [];
      });
      var localDatabases = databases.filter(function (database) { return database._id !== next._id; }).concat([next]);
      var context = ctx.referenceContext(localDatabases, next);
      added.forEach(function (item, position) {
        var values = convert(item.row.values, current.schema, databases, next, planned);
        next.records[current.records.length + position].values = validation.normalizeValues(values, current.schema, context, null);
      });
      next.importMappings = mappings(next).concat(added.map(function (item) { return item.mapping; }));
      next.revision++; next.updatedAt = at; next.updatedBy = who; next.projectionPending = true; next.projectionError = null;
      result.databaseRevision = next.revision;
      ctx.remember(next, operationId, fingerprint, result);
      await ctx.replace(current, next, args);
      return ctx.finish(next, result, who);
    }); });
  }
  function readImportProgress(args) {
    var namespace = input.sourceText(args.sourceNamespace, "Source namespace", 120);
    var offset = args.offset === undefined ? 0 : args.offset; var limit = args.limit === undefined ? 100 : args.limit;
    if (!Number.isInteger(offset) || offset < 0 || !Number.isInteger(limit) || limit < 1 || limit > 100) throw ctx.error("Import progress requires a nonnegative integer offset and limit between 1 and 100.", "INVALID");
    return ctx.load(args.databaseId).then(function (database) {
      var rows = mappings(database).filter(function (mapping) { return mapping.sourceNamespace === namespace; });
      rows.sort(function (a, b) { return a.sourceKey === b.sourceKey ? 0 : a.sourceKey < b.sourceKey ? -1 : 1; });
      return { databaseId: database._id, databaseRevision: database.revision, sourceNamespace: namespace, total: rows.length, mappings: ctx.clone(rows.slice(offset, offset + limit)), nextOffset: offset + limit < rows.length ? offset + limit : null };
    });
  }
  return { importRecords: importRecords, readImportProgress: readImportProgress };
}

module.exports = { attachKnowledgeDbImport: attachKnowledgeDbImport, MAX_BATCH_ROWS: input.MAX_BATCH_ROWS, MAX_BATCH_BYTES: input.MAX_BATCH_BYTES };