lib/knowledge-import.js
Rama main · réplica de lectura del repositorio original.
// Import-keyed projection over the Clay-wide knowledge record backend.
//
// A migration and a live write-through bridge both copy legacy content in
// repeatedly: on every restart, after a partial run, after every mutation, and
// potentially from two daemons at once. This layer makes that safe by
// addressing every imported record with a stable import key derived from the
// source, so re-importing the same source is a no-op rather than a duplicate.
//
// Change rule, applied by the caller through the key it chooses:
// - File-addressed sources use a key derived from the file name. Re-importing
// changed content revises the existing record, keeping the prior revision
// and its provenance.
// - Line-addressed sources (append-only journals) fold the line's content
// hash into the key. An edited line is therefore a new, deterministic
// import rather than a revision, which matches how those journals are
// actually written.
//
// Content is never truncated. A source larger than one record is stored as
// content-addressed chunks and reassembled exactly, with the whole-source hash
// verified on read. A caller that cannot read a source back completely is told
// so rather than handed a silently partial string.
var crypto = require("crypto");
var recordStore = require("./knowledge-record-store");
var IMPORT_VERSION = 2;
// Worst-case JSON/UTF-8 expansion of an 8000 character slice stays well inside
// the record store's 64KB single-record limit.
var CHUNK_CHARS = 8000;
var MAX_CONTENT_CHARS = CHUNK_CHARS;
var MAX_NAME_CHARS = 200;
var MAX_CHUNKS = 8192;
var NUL = String.fromCharCode(0);
var CONTROL_PATTERN = new RegExp("[\\u0000-\\u001f\\u007f]+", "g");
function contentHash(value) {
return crypto.createHash("sha256").update(String(value == null ? "" : value), "utf8").digest("hex").substring(0, 32);
}
function cleanLine(value, max) {
if (typeof value !== "string") return "";
var text = value.replace(CONTROL_PATTERN, " ").replace(/\s+/g, " ").trim();
return text.length > max ? text.substring(0, max) : text;
}
// Content is stored verbatim. Every character survives a JSONL round trip,
// including NUL, which JSON.stringify escapes as a six-character sequence and JSON.parse restores
// exactly; it is not a newline, so it cannot split a record either. Removing
// any character here would break reassembly against the source hash.
function exactContent(value) {
return typeof value === "string" ? value : "";
}
function splitChunks(content) {
var chunks = [];
for (var offset = 0; offset < content.length; offset += CHUNK_CHARS) {
chunks.push(content.substring(offset, offset + CHUNK_CHARS));
}
return chunks;
}
function chunkKey(importKey, position) {
return importKey + "#c" + position;
}
function createKnowledgeImporter(opts) {
var options = opts || {};
var scopeId = options.scopeId;
var store = recordStore.createRecordStore({ scopeId: scopeId, baseDir: options.baseDir });
function refFor(rootId) {
var digest = crypto.createHash("sha256").update(scopeId + NUL + String(rootId)).digest("base64url");
return "know:" + digest.substring(0, 24);
}
// Incrementally maintained so importing N sources stays O(N) rather than
// re-folding the whole scope once per source. The bridge runs on every
// mutation and the migration on every startup, so this must stay cheap.
var indexed = null;
function resetIndex() {
indexed = { byKey: new Map(), byRoot: new Map(), duplicates: 0, consumed: 0 };
}
function foldRecord(record) {
if (record.op === "create" && record.importKey) {
// First create for an import key wins. A duplicate that slipped past the
// run lock is folded away rather than double-counted, so the projection
// stays correct even if the lock ever fails.
if (indexed.byKey.has(record.importKey)) {
indexed.duplicates++;
return;
}
var state = {
importKey: record.importKey,
rootId: record.id,
ref: refFor(record.id),
sourceHash: record.sourceHash || null,
contentHash: record.contentHash || record.sourceHash || null,
kind: record.kind || null,
name: record.name || null,
chunked: record.chunked === true,
chunkCount: record.chunkCount || 0,
deleted: false,
revisions: 1,
importedAt: record.at || 0,
updatedAt: record.at || 0,
};
indexed.byKey.set(record.importKey, state);
indexed.byRoot.set(record.id, state);
return;
}
if (!record.rootId) return;
var target = indexed.byRoot.get(record.rootId);
if (!target) return;
if (record.op === "update") {
if (record.sourceHash) target.sourceHash = record.sourceHash;
if (record.contentHash) target.contentHash = record.contentHash;
if (record.kind) target.kind = record.kind;
if (record.name) target.name = record.name;
target.chunked = record.chunked === true;
target.chunkCount = record.chunkCount || 0;
// An update always asserts that this content is current, so it revives a
// record whose source came back after a tombstone.
target.deleted = false;
} else if (record.op === "delete") {
target.deleted = true;
} else {
return;
}
target.revisions++;
target.updatedAt = record.at || target.updatedAt;
}
function index() {
var records = store.all();
if (!indexed || records.length < indexed.consumed) resetIndex();
for (var i = indexed.consumed; i < records.length; i++) foldRecord(records[i]);
indexed.consumed = records.length;
return { byKey: indexed.byKey, duplicates: indexed.duplicates };
}
// Folded content for every live import key. Used only for reassembling a
// chunked source, which is rare, so it is computed on demand.
function foldContents() {
var records = store.all();
var byRoot = new Map();
var byKey = new Map();
for (var i = 0; i < records.length; i++) {
var record = records[i];
if (record.op === "create" && record.importKey) {
if (byKey.has(record.importKey)) continue;
var state = { content: record.content || "", deleted: false };
byKey.set(record.importKey, state);
byRoot.set(record.id, state);
continue;
}
if (!record.rootId) continue;
var target = byRoot.get(record.rootId);
if (!target) continue;
if (record.op === "update") {
if (typeof record.content === "string") target.content = record.content;
target.deleted = false;
} else if (record.op === "delete") {
target.deleted = true;
}
}
return byKey;
}
// Exact reassembly. Returns complete:false rather than a partial string when
// a chunk is missing, tombstoned, or the whole-source hash does not verify.
function readContent(entry) {
if (!entry) return { complete: false, content: "", reason: "not-found" };
if (!entry.chunked) {
// Verified on the inline path too, so "complete" means the same thing
// whether or not a source happened to need chunking.
var inline = entry.content || "";
if (entry.contentHash && contentHash(inline) !== entry.contentHash) {
return { complete: false, content: "", reason: "hash-mismatch" };
}
return { complete: true, content: inline };
}
var contents = foldContents();
var parts = [];
for (var i = 0; i < entry.chunkCount; i++) {
var part = contents.get(chunkKey(entry.importKey, i));
if (!part || part.deleted) return { complete: false, content: "", reason: "missing-chunk", chunkIndex: i };
parts.push(part.content);
}
var joined = parts.join("");
if (entry.contentHash && contentHash(joined) !== entry.contentHash) {
return { complete: false, content: "", reason: "hash-mismatch" };
}
return { complete: true, content: joined };
}
function append(payload) {
return store.append(payload);
}
// Chunks are written before the parent record. A failure partway therefore
// leaves orphan chunks and no parent, so the source is not reported as
// imported and the next run retries it. Chunks are content-addressed, so the
// retry re-uses whatever already landed.
function writeChunks(importKey, chunks, entry, at) {
var existingIndex = index().byKey;
for (var i = 0; i < chunks.length; i++) {
var key = chunkKey(importKey, i);
var hash = contentHash(chunks[i]);
var current = existingIndex.get(key);
if (current && !current.deleted && current.sourceHash === hash) continue;
var base = {
op: current ? "update" : "create",
scope: scopeId,
importKey: key,
sourceHash: hash,
contentHash: hash,
kind: "chunk",
name: entry.name || "",
chunkOf: importKey,
chunkIndex: i,
content: chunks[i],
importVersion: IMPORT_VERSION,
at: at,
};
if (current) {
base.rootId = current.rootId;
} else {
var id = recordStore.newRecordId();
base.id = id;
base.rootId = id;
}
append(base);
}
}
// Import one source unit. Returns what actually happened so a migration or a
// bridge can report truthfully instead of assuming.
function importRecord(input) {
var entry = input || {};
if (!entry.importKey || typeof entry.importKey !== "string") throw new Error("An import key is required.");
var content = exactContent(entry.content);
// Two distinct hashes, deliberately.
// sourceHash - the caller's fingerprint of the legacy source, used for
// change detection and for line-addressed import keys.
// storedHash - always derived here from the exact bytes being written,
// so reassembly verifies against what was actually stored
// rather than against something the caller asserted.
var storedHash = contentHash(content);
var hash = entry.sourceHash || storedHash;
var existing = index().byKey.get(entry.importKey) || null;
if (existing && !existing.deleted && existing.sourceHash === hash) {
return { action: "unchanged", ref: existing.ref, importKey: entry.importKey };
}
var at = Number.isFinite(entry.at) ? entry.at : Date.now();
var chunks = content.length > MAX_CONTENT_CHARS ? splitChunks(content) : null;
if (chunks) {
if (chunks.length > MAX_CHUNKS) {
throw new Error("Source exceeds the maximum representable size: " + entry.importKey);
}
writeChunks(entry.importKey, chunks, entry, at);
}
var payload = {
op: existing ? "update" : "create",
scope: scopeId,
importKey: entry.importKey,
sourceHash: hash,
contentHash: storedHash,
kind: entry.kind || "knowledge",
name: cleanLine(entry.name || (existing && existing.name) || "", MAX_NAME_CHARS),
chunked: !!chunks,
chunkCount: chunks ? chunks.length : 0,
content: chunks ? "" : content,
source: entry.source || null,
actor: entry.actor || null,
importVersion: IMPORT_VERSION,
at: at,
};
if (existing) {
payload.rootId = existing.rootId;
append(payload);
return { action: existing.deleted ? "revived" : "revised", ref: existing.ref, importKey: entry.importKey };
}
var newId = recordStore.newRecordId();
payload.id = newId;
payload.rootId = newId;
append(payload);
return { action: "created", ref: refFor(newId), importKey: entry.importKey };
}
// Logical removal. The record and its history stay; the projection stops
// returning it, so a memory deleted in the legacy store is not left active
// in the new backend.
function removeRecord(importKey, options2) {
var settings = options2 || {};
var existing = index().byKey.get(importKey) || null;
if (!existing) return { action: "absent", importKey: importKey };
if (existing.deleted) return { action: "already-deleted", ref: existing.ref, importKey: importKey };
append({
rootId: existing.rootId,
op: "delete",
scope: scopeId,
importKey: importKey,
kind: existing.kind,
name: existing.name,
source: settings.source || null,
actor: settings.actor || null,
importVersion: IMPORT_VERSION,
at: Number.isFinite(settings.at) ? settings.at : Date.now(),
});
return { action: "deleted", ref: existing.ref, importKey: importKey };
}
// Read projection. Chunk records are internal and never surface here; use
// readContent to reassemble a chunked entry.
function entries(options2) {
var settings = options2 || {};
var records = store.all();
var byRoot = new Map();
var byKey = new Map();
var order = [];
var i;
for (i = 0; i < records.length; i++) {
var record = records[i];
if (record.op !== "create" || !record.importKey) continue;
if (byKey.has(record.importKey)) continue;
var entry = {
ref: refFor(record.id),
importKey: record.importKey,
kind: record.kind || null,
name: record.name || null,
content: record.content || "",
chunked: record.chunked === true,
chunkCount: record.chunkCount || 0,
contentHash: record.contentHash || record.sourceHash || null,
source: record.source || null,
actor: record.actor || null,
deleted: false,
deletedAt: null,
deletedBy: null,
importedAt: record.at || 0,
updatedAt: record.at || 0,
revisions: 1,
};
byKey.set(record.importKey, entry);
byRoot.set(record.id, entry);
order.push(entry);
}
for (i = 0; i < records.length; i++) {
var revision = records[i];
if (revision.op === "create" || !revision.rootId) continue;
var target = byRoot.get(revision.rootId);
if (!target) continue;
if (revision.op === "delete") {
// Original authorship is preserved on `actor`; who removed it is
// recorded separately so a tombstone is traceable too.
target.deleted = true;
target.deletedAt = revision.at || 0;
target.deletedBy = revision.actor || null;
} else if (revision.op === "update") {
if (typeof revision.content === "string") target.content = revision.content;
if (revision.name) target.name = revision.name;
if (revision.kind) target.kind = revision.kind;
if (revision.source) target.source = revision.source;
if (revision.actor) target.actor = revision.actor;
if (revision.contentHash) target.contentHash = revision.contentHash;
target.chunked = revision.chunked === true;
target.chunkCount = revision.chunkCount || 0;
target.deleted = false;
target.deletedAt = null;
target.deletedBy = null;
} else {
continue;
}
target.updatedAt = revision.at || target.updatedAt;
target.revisions++;
}
var out = [];
for (i = 0; i < order.length; i++) {
if (order[i].kind === "chunk" && settings.includeChunks !== true) continue;
if (order[i].deleted && settings.includeDeleted !== true) continue;
out.push(order[i]);
}
return out;
}
function stats() {
var indexedNow = index();
var base = store.stats();
var live = 0;
var deleted = 0;
var chunks = 0;
indexedNow.byKey.forEach(function (state) {
if (state.kind === "chunk") { chunks++; return; }
if (state.deleted) deleted++;
else live++;
});
return {
scopeId: scopeId,
filePath: base.filePath,
records: base.records,
skippedRecords: base.skipped,
imported: live,
deleted: deleted,
chunks: chunks,
duplicates: indexedNow.duplicates,
};
}
return {
scopeId: scopeId,
filePath: store.filePath,
index: index,
importRecord: importRecord,
removeRecord: removeRecord,
readContent: readContent,
entries: entries,
stats: stats,
};
}
module.exports = {
IMPORT_VERSION: IMPORT_VERSION,
CHUNK_CHARS: CHUNK_CHARS,
MAX_CONTENT_CHARS: MAX_CONTENT_CHARS,
MAX_CHUNKS: MAX_CHUNKS,
contentHash: contentHash,
chunkKey: chunkKey,
splitChunks: splitChunks,
createKnowledgeImporter: createKnowledgeImporter,
};