poller.js

// sync-files/poller.js
import path from 'path';
import { fileURLToPath } from 'url';
import dotenv from 'dotenv';
import fs from 'fs';
import { Client } from '@notionhq/client';

const __filename = fileURLToPath(import.meta.url);
const __dirname  = path.dirname(__filename);

// --- Logger-Bridge (vom Controller setzbar)
let logger = (..._args) => {};
export function setLogger(fn){ logger = typeof fn === 'function' ? fn : (()=>{}); }

// .env laden (Root bevorzugt, sonst lokal neben Datei)
const rootEnv = path.resolve(__dirname, '..', '.env');
const localEnv = path.join(__dirname, '.env');
dotenv.config({ path: fs.existsSync(rootEnv) ? rootEnv : (fs.existsSync(localEnv) ? localEnv : undefined) });

// ENV-Guards
if (!process.env.NOTION_TOKEN)    throw new Error('NOTION_TOKEN fehlt');
if (!process.env.ALL_TASKS_DB_ID) throw new Error('ALL_TASKS_DB_ID fehlt');
if (!process.env.MODULE_DB_IDS)   throw new Error('MODULE_DB_IDS fehlt');

// Performance-Parameter (steuerbar via .env)
const MODULE_DB_IDS = process.env.MODULE_DB_IDS.split(',').map(s => s.trim()).filter(Boolean);
const DS_CONCURRENCY = Math.max(1, parseInt(process.env.DS_CONCURRENCY || '4', 10));
const PAGE_SIZE      = Math.min(100, Math.max(10, parseInt(process.env.PAGE_SIZE || '100', 10)));
const HYDRATE_EVERY_RUN = String(process.env.HYDRATE_EVERY_RUN || 'false').toLowerCase() === 'true';

const notion = new Client({
  auth: process.env.NOTION_TOKEN,
  notionVersion: '2025-09-03',
  timeoutMs: 90_000,
});

// ---- Persistenz: über MAPPING_JSON ----
let MAP_FILE = process.env.MAPPING_JSON ?? path.resolve(__dirname, 'mapping.json');
if (!fs.existsSync(MAP_FILE)) {
  fs.writeFileSync(MAP_FILE, JSON.stringify({ map: {}, lastSync: null }, null, 2), 'utf8');
}
const DELETED_FILE = MAP_FILE.replace(/mapping\.json$/i, 'deleted.json');
const QUEUE_FILE   = MAP_FILE.replace(/mapping\.json$/i, 'queue.json');

function readJsonSafe(file, fb){ try { return JSON.parse(fs.readFileSync(file,'utf8')); } catch { return fb; } }
function writeJsonAtomic(file, data){
  const tmp = file + '.tmp', bak = file + '.bak';
  try { if (fs.existsSync(file)) fs.copyFileSync(file, bak); } catch {}
  fs.writeFileSync(tmp, JSON.stringify(data, null, 2), 'utf8');
  fs.renameSync(tmp, file);
}

let store = { map: {}, lastSync: null };
if (fs.existsSync(MAP_FILE)) { try { store = JSON.parse(fs.readFileSync(MAP_FILE,'utf8')) || store; } catch {} }
const saveStore   = () => writeJsonAtomic(MAP_FILE, store);

let deleted = new Set();
if (fs.existsSync(DELETED_FILE)) {
  try { const raw = JSON.parse(fs.readFileSync(DELETED_FILE,'utf8')); if (Array.isArray(raw)) deleted = new Set(raw); } catch {}
} else {
  fs.writeFileSync(DELETED_FILE, JSON.stringify([], null, 2), 'utf8');
}
const saveDeleted = () => writeJsonAtomic(DELETED_FILE, [...deleted]);

let queue = fs.existsSync(QUEUE_FILE) ? readJsonSafe(QUEUE_FILE, []) : [];
const saveQueue = () => writeJsonAtomic(QUEUE_FILE, queue);

let processing = false;
let stopQueue  = false;
function enqueue(op){ queue.push({ id: Date.now() + '-' + Math.random().toString(36).slice(2), ...op }); saveQueue(); }
export function getPendingSize(){ return queue.length + (processing ? 1 : 0); }

// ---- Feldnamen
const NAMES = {
  sourceTitle: 'Topic',
  sourceDateCandidates: ['Date','Dates'],
  targetTitle: 'Topic',
  targetDate:  'Date',
  sourceIdText: 'SourceId',
};

const SAFETY_WINDOW_MS = 120_000;
const sleep = (ms) => new Promise(r => setTimeout(r, ms));

// ---- Retry
async function withRetry(fn, { tries = 5, baseDelay = 400 } = {}) {
  let attempt = 0, lastErr;
  while (attempt < tries) {
    try { return await fn(); }
    catch (e) {
      lastErr = e;
      const status = e?.status || e?.code;
      if (status && status !== 429 && status < 500) break;
      await sleep(baseDelay * Math.pow(2, attempt));
      attempt++;
    }
  }
  throw lastErr;
}

// ---- Helpers
function getTitle(props) {
  const t = props?.[NAMES.sourceTitle]?.title;
  return (t && t.length ? t.map(x => x.plain_text).join('') : '').trim() || 'Untitled';
}
function getDate(props) {
  for (const name of NAMES.sourceDateCandidates) {
    const d = props?.[name]?.date?.start;
    if (d) return d;
  }
  return null;
}
async function getPage(pageId) {
  return withRetry(() => notion.pages.retrieve({ page_id: pageId }));
}

// ---- Queries
async function fetchChangedPages(dataSourceId, sinceISO) {
  const since = sinceISO
    ? new Date(Math.max(0, Date.parse(sinceISO) - SAFETY_WINDOW_MS)).toISOString()
    : undefined;

  const filter = since
    ? { timestamp: 'last_edited_time', last_edited_time: { after: since } }
    : undefined;

  let cursor; const out = [];
  do {
    if (!notion.dataSources?.query) {
      throw new Error('Dein @notionhq/client unterstützt dataSources.* nicht. Bitte @notionhq/client@latest installieren.');
    }
    const resp = await withRetry(() => notion.dataSources.query({
      data_source_id: dataSourceId,
      start_cursor: cursor,
      filter,
      sorts: [{ timestamp: 'last_edited_time', direction: 'ascending' }],
      page_size: PAGE_SIZE,
    }));
    out.push(...resp.results);
    cursor = resp.has_more ? resp.next_cursor : undefined;
  } while (cursor);
  return out;
}

let hydratedOnce = false;

async function hydrateMapFromAllTasks() {
  let cursor;
  do {
    const resp = await withRetry(() => notion.dataSources.query({
      data_source_id: process.env.ALL_TASKS_DB_ID,
      start_cursor: cursor,
      page_size: PAGE_SIZE,
    }));
    for (const pg of resp.results) {
      const sid = pg.properties?.[NAMES.sourceIdText]?.rich_text?.[0]?.plain_text;
      if (sid && !store.map[sid]) store.map[sid] = pg.id;
    }
    cursor = resp.has_more ? resp.next_cursor : undefined;
  } while (cursor);
}

function buildTargetPropsFromSource(sourceProps) {
  const title = getTitle(sourceProps);
  const date  = getDate(sourceProps);
  const props = {
    [NAMES.targetTitle]: { title: [{ type: 'text', text: { content: title } }] },
  };
  if (date) props[NAMES.targetDate] = { date: { start: date } };
  return props;
}

// ---- CRUD All Tasks
async function trashPage(pageId, makeArchived = true, tries = 5) {
  try {
    await withRetry(() => notion.pages.update({ page_id: pageId, archived: makeArchived }));
  } catch (e) {
    const msg = e?.body?.message || e?.message || '';
    if (!(makeArchived && /archived/i.test(msg))) throw e;
  }
  for (let i = 0; i < tries; i++) {
    const pg = await getPage(pageId).catch(() => null);
    if (!pg) break;
    const ok = makeArchived ? !!pg.in_trash : !pg.in_trash;
    if (ok) return pg;
    await sleep(500);
  }
  return getPage(pageId).catch(() => null);
}

async function archiveInAllTasks(targetPageId, sourceIdToRemember) {
  try {
    const pg = await trashPage(targetPageId, true);
    logger(`Trash AllTasks: ${targetPageId} (in_trash: ${pg?.in_trash})`);
    if (sourceIdToRemember) { deleted.add(sourceIdToRemember); saveDeleted(); }
    delete store.map[sourceIdToRemember];
    saveStore();
  } catch (e) {
    logger(`ERROR trash target: ${e?.body?.message || e?.message || e}`);
  }
}
async function archiveInSource(sourcePageId) {
  try {
    const pg = await trashPage(sourcePageId, true);
    logger(`Trash Source: ${sourcePageId} (in_trash: ${pg?.in_trash})`);
    deleted.add(sourcePageId); 
    saveDeleted();
  } catch (e) {
    if (e.code === 'object_not_found') {
      logger(`Source ${sourcePageId} already gone`);
    } else {
      logger(`ERROR trash source: ${e?.body?.message || e?.message || e}`);
    }
  }
}

async function createInAllTasks(sourcePage) {
  if (store.map[sourcePage.id]) return null;

  const dsId = process.env.ALL_TASKS_DB_ID;
  let cursor, foundExisting = null;
  do {
    const resp = await withRetry(() => notion.dataSources.query({
      data_source_id: dsId,
      start_cursor: cursor,
      page_size: PAGE_SIZE,
    }));
    for (const pg of resp.results) {
      const sid = pg.properties?.[NAMES.sourceIdText]?.rich_text?.[0]?.plain_text;
      if (sid === sourcePage.id) { foundExisting = pg.id; break; }
    }
    cursor = foundExisting ? undefined : (resp.has_more ? resp.next_cursor : undefined);
  } while (cursor);

  if (foundExisting) {
    store.map[sourcePage.id] = foundExisting; 
    saveStore();
    return foundExisting;
  }

  const props = buildTargetPropsFromSource(sourcePage.properties);
  props[NAMES.sourceIdText] = { rich_text: [{ type: 'text', text: { content: sourcePage.id } }] };
  const res = await withRetry(() => notion.pages.create({
    parent: { data_source_id: process.env.ALL_TASKS_DB_ID },
    properties: props,
  }));
  store.map[sourcePage.id] = res.id;
  saveStore();
  return res.id;
}

async function updateInAllTasks(targetPageId, sourcePage) {
  const srcUpdated = new Date(sourcePage.last_edited_time);
  const tgtFull = await getPage(targetPageId);
  if (new Date(tgtFull.last_edited_time) > srcUpdated) return false;
  const props = buildTargetPropsFromSource(sourcePage.properties);
  await withRetry(() => notion.pages.update({ page_id: targetPageId, properties: props }));
  return true;
}

// ---- Queue
async function execOp(op){
  if (op.type === 'create') {
    if (store.map[op.sourceId]) return { created:0 };

    const dsId = process.env.ALL_TASKS_DB_ID;
    let cursor, foundExisting = null;
    do {
      const resp = await withRetry(() => notion.dataSources.query({
        data_source_id: dsId,
        start_cursor: cursor,
        page_size: PAGE_SIZE,
      }));
      for (const pg of resp.results) {
        const sid = pg.properties?.[NAMES.sourceIdText]?.rich_text?.[0]?.plain_text;
        if (sid === op.sourceId) { foundExisting = pg.id; break; }
      }
      cursor = foundExisting ? undefined : (resp.has_more ? resp.next_cursor : undefined);
    } while (cursor);

    if (foundExisting) {
      store.map[op.sourceId] = foundExisting; saveStore();
      return { created:0 };
    }

    const res = await withRetry(() => notion.pages.create({
      parent: { data_source_id: process.env.ALL_TASKS_DB_ID },
      properties: {
        ...op.props,
        [NAMES.sourceIdText]: { rich_text: [{ type: 'text', text: { content: op.sourceId } }] },
      },
    }));
    store.map[op.sourceId] = res.id; saveStore();
    return { created:1 };
  }

  if (op.type === 'archive-target') {
    if (op.targetId) {
      const tgt = await getPage(op.targetId).catch(() => null);
      if (!tgt || tgt.in_trash || tgt.archived) {
        delete store.map[op.sourceId];
        saveStore();
        return { archived: 1 };
      }
      await trashPage(op.targetId, true);
      delete store.map[op.sourceId];
      saveStore();
      return { archived:1 };
    }
    return { archived:0 };
  }

  if (op.type === 'archive-source') {
    await trashPage(op.sourceId, true);
    deleted.add(op.sourceId); saveDeleted();
    return { archived:1 };
  }

  return {};
}

async function processQueue({ sliceOnly = false } = {}){
  processing = true;
  const delta = { created:0, updated:0, archived:0 };
  try {
    while (queue.length && !stopQueue) {
      const op = queue[0];
      const res = await execOp(op).catch(err => {
        logger(`[QUEUE] op failed: ${err?.body?.message || err?.message || err}`);
        return {};
      });
      if (res && Object.keys(res).length) {
        queue.shift(); saveQueue();
        delta.created  += res.created  || 0;
        delta.archived += res.archived || 0;
      } else {
        break;
      }
      if (sliceOnly) break;
    }
  } finally { processing = false; }
  return delta;
}

export async function drainPending({ timeoutMs = 15000, sliceOnly = false } = {}){
  stopQueue = false;
  const t0 = Date.now();
  let totals = { created:0, updated:0, archived:0 };
  while ((queue.length || processing) && Date.now()-t0 < timeoutMs) {
    const d = await processQueue({ sliceOnly });
    totals.created  += d.created;
    totals.archived += d.archived;
    if (sliceOnly) break;
  }
  return totals;
}

// ---- Syncs
async function syncDeletionsFromAllTasks() {
  let cursor, archivedCount = 0;
  do {
    const resp = await withRetry(() => notion.dataSources.query({
      data_source_id: process.env.ALL_TASKS_DB_ID,
      start_cursor: cursor,
      page_size: PAGE_SIZE,
    }));
    for (const pg of resp.results) {
      const sid = pg.properties?.[NAMES.sourceIdText]?.rich_text?.[0]?.plain_text;
      if (!sid) continue;
      if (pg.in_trash || pg.archived) {
        await archiveInSource(sid);
        delete store.map[sid];
        archivedCount++;
      }
    }
    cursor = resp.has_more ? resp.next_cursor : undefined;
  } while (cursor);
  return archivedCount; // nicht für UI gezählt
}

async function archiveTargetsWhoseSourcesMissing() {
  let queued = 0;
  const entries = Object.entries(store.map);
  for (const [sourceId, targetId] of entries) {
    let src = null;
    try {
      src = await getPage(sourceId);
    } catch (e) {
      if (e?.code === 'object_not_found' || e?.status === 404) {
        enqueue({ type: 'archive-target', targetId, sourceId });
        queued++;
        continue;
      } else {
        logger(`Kurzzeitig kein Zugriff auf eine Quelle – wird automatisch erneut versucht.`);
        continue;
      }
    }
    if (src.in_trash || src.archived) {
      enqueue({ type: 'archive-target', targetId, sourceId });
      queued++; continue;
    }
    const parentOk = (src?.parent?.type === 'data_source_id') &&
                     MODULE_DB_IDS.includes(src.parent.data_source_id);
    if (!parentOk) {
      enqueue({ type: 'archive-target', targetId, sourceId });
      queued++; continue;
    }
  }
  if (queued) logger(`Löschung(en) im Bereich „All Tasks“ wird/werden angestoßen.`);
  return queued;
}

// ---- Public API
export async function verifyAccess() {
  await withRetry(() => notion.dataSources.retrieve({ data_source_id: process.env.ALL_TASKS_DB_ID }));
  for (const id of MODULE_DB_IDS) {
    await withRetry(() => notion.dataSources.retrieve({ data_source_id: id }));
  }
  logger('Access OK');
}

async function processOneDataSource(dsId, since, counters) {
  const pages = await fetchChangedPages(dsId, since);
  for (const p of pages) {
    const targetId = store.map[p.id];
    if (p.in_trash || p.archived) {
      if (targetId) enqueue({ type:'archive-target', targetId, sourceId: p.id });
      else enqueue({ type:'archive-source', sourceId: p.id });
      continue;
    }
    if (!p.in_trash && !p.archived && deleted.has(p.id)) { deleted.delete(p.id); saveDeleted(); }
    if (!targetId) {
      if (deleted.has(p.id)) continue;
      const props = buildTargetPropsFromSource(p.properties);
      enqueue({ type:'create', sourceId: p.id, props });
    } else {
      const tgt = await getPage(targetId).catch(() => null);
      if (tgt?.archived || tgt?.in_trash) { delete store.map[p.id]; continue; }
      const changed = await updateInAllTasks(targetId, p);
      if (changed) counters.updated++;
    }
  }
}

export async function runOnce() {
  if (!hydratedOnce || HYDRATE_EVERY_RUN || Object.keys(store.map).length === 0) {
    await hydrateMapFromAllTasks();
    hydratedOnce = true;
  }
  await syncDeletionsFromAllTasks();

  let created = 0, updated = 0, archived = 0;
  const since = store.lastSync;

  // Parallel über Quellen (limitiert)
  const ids = MODULE_DB_IDS.slice();
  const groups = []; for (let i=0;i<DS_CONCURRENCY;i++) groups.push([]);
  for (let i=0;i<ids.length;i++) groups[i % DS_CONCURRENCY].push(ids[i]);

  const perRunCounters = { updated: 0 };
  await Promise.all(groups.map(group => (async () => {
    for (const id of group) await processOneDataSource(id, since, perRunCounters);
  })()));
  updated += perRunCounters.updated;

  await archiveTargetsWhoseSourcesMissing();

  const d = await drainPending({ timeoutMs: 30_000, sliceOnly: false });
  created  += d.created  || 0;
  archived += d.archived || 0;

  store.lastSync = new Date().toISOString();
  saveStore();

  return { changed: (created + updated + archived) > 0, stats: { created, updated, archived } };
}