99 lines
3.9 KiB
JavaScript
99 lines
3.9 KiB
JavaScript
const os = require('os');
|
|
const Database = require('better-sqlite3');
|
|
const { initAutonomySchema } = require('../src/autonomy/schema');
|
|
const { enqueueJob, leaseNextJob, completeJob, failJob } = require('../src/autonomy/jobs');
|
|
|
|
function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); }
|
|
|
|
function reconcileArchiveBatch(archiveDb, intelligenceDb, batchSize = 250) {
|
|
const cursor = intelligenceDb.prepare("SELECT value FROM autonomy_cursors WHERE key = 'archive_reconcile'").get();
|
|
const afterId = cursor ? cursor.value : 0;
|
|
const rows = archiveDb.prepare(`
|
|
SELECT id, event_id, ingested_at, content, has_embedding
|
|
FROM articles
|
|
WHERE id > ?
|
|
ORDER BY id ASC
|
|
LIMIT ?
|
|
`).all(afterId, batchSize);
|
|
if (!rows.length) {
|
|
intelligenceDb.prepare(`
|
|
INSERT INTO autonomy_cursors(key, value) VALUES ('archive_reconcile', 0)
|
|
ON CONFLICT(key) DO UPDATE SET value = 0, updated_at = datetime('now')
|
|
`).run();
|
|
return { scanned: 0, nextCursor: 0, reset: true };
|
|
}
|
|
const enqueue = intelligenceDb.transaction(() => {
|
|
for (const row of rows) {
|
|
const isLive = row.ingested_at && Date.now() - Date.parse(row.ingested_at) <= 48 * 60 * 60 * 1000;
|
|
const readyForIntelligence = row.event_id && row.content && row.has_embedding;
|
|
if (readyForIntelligence) {
|
|
enqueueJob(intelligenceDb, {
|
|
jobType: 'coordinator_event',
|
|
lane: isLive ? 'live' : 'historical',
|
|
priority: isLive ? 100 : 10,
|
|
entityType: 'event',
|
|
entityId: row.event_id,
|
|
idempotencyKey: `coordinator_event:${row.event_id}`,
|
|
});
|
|
}
|
|
}
|
|
intelligenceDb.prepare(`
|
|
INSERT INTO autonomy_cursors(key, value) VALUES ('archive_reconcile', ?)
|
|
ON CONFLICT(key) DO UPDATE SET value = excluded.value, updated_at = datetime('now')
|
|
`).run(rows[rows.length - 1].id);
|
|
});
|
|
enqueue();
|
|
return { scanned: rows.length, nextCursor: rows[rows.length - 1].id, reset: false };
|
|
}
|
|
|
|
function reconcileLiveBatch(archiveDb, intelligenceDb, batchSize = 250) {
|
|
const rows = archiveDb.prepare(`
|
|
SELECT id, event_id, ingested_at, content, has_embedding
|
|
FROM articles
|
|
WHERE ingested_at >= datetime('now', '-48 hours')
|
|
ORDER BY ingested_at DESC, id DESC
|
|
LIMIT ?
|
|
`).all(batchSize);
|
|
let queued = 0;
|
|
for (const row of rows) {
|
|
if (!row.event_id || !row.content || !row.has_embedding) continue;
|
|
const result = enqueueJob(intelligenceDb, {
|
|
jobType: 'coordinator_event', lane: 'live', priority: 100,
|
|
entityType: 'event', entityId: row.event_id,
|
|
idempotencyKey: `coordinator_event:${row.event_id}`,
|
|
});
|
|
if (result.inserted) queued++;
|
|
}
|
|
return { scanned: rows.length, queued };
|
|
}
|
|
|
|
async function runAutonomyWorker({ archivePath, intelligencePath, workerId = `autonomy-${os.hostname()}-${process.pid}`, pollMs = 1000 } = {}) {
|
|
const archiveDb = new Database(archivePath, { readonly: true });
|
|
const intelligenceDb = new Database(intelligencePath);
|
|
intelligenceDb.pragma('journal_mode = WAL');
|
|
initAutonomySchema(intelligenceDb);
|
|
|
|
while (true) {
|
|
const job = leaseNextJob(intelligenceDb, workerId, 120);
|
|
if (!job) { await sleep(pollMs); continue; }
|
|
try {
|
|
if (job.job_type === 'reconcile_archive') {
|
|
reconcileLiveBatch(archiveDb, intelligenceDb);
|
|
reconcileArchiveBatch(archiveDb, intelligenceDb);
|
|
// Keep the reconciler alive as a bounded maintenance loop.
|
|
enqueueJob(intelligenceDb, {
|
|
jobType: 'reconcile_archive', lane: 'maintenance', priority: 100,
|
|
entityType: 'archive', entityId: 'archive',
|
|
idempotencyKey: `reconcile_archive:${Date.now()}`,
|
|
});
|
|
}
|
|
completeJob(intelligenceDb, job.id, workerId);
|
|
} catch (error) {
|
|
failJob(intelligenceDb, job.id, workerId, error);
|
|
}
|
|
await sleep(pollMs);
|
|
}
|
|
}
|
|
|
|
module.exports = { reconcileArchiveBatch, reconcileLiveBatch, runAutonomyWorker };
|