Persist notification email batches with leased idempotent delivery

server/notify-store.mjs

@@ -1,4 +1,4 @@
-import {createHash} from 'node:crypto';
+import {createHash,randomUUID} from 'node:crypto';
 /**
  * Otomatik e-posta bildirimi abonelikleri (SQLite).
  * Kullanıcı kendi filtre kriterlerini kaydeder; zamanlayıcı yeni eşleşen
@@ -39,6 +39,8 @@ CREATE TABLE IF NOT EXISTS abonelikler (
   UNIQUE(email, criteria_key)
 );
 CREATE INDEX IF NOT EXISTS idx_abone_email ON abonelikler(email);
+CREATE TABLE IF NOT EXISTS notification_outbox(id TEXT PRIMARY KEY,subscription_id INTEGER NOT NULL,payload TEXT NOT NULL,tender_ids TEXT NOT NULL,state TEXT NOT NULL,first_attempt INTEGER,lease TEXT,lease_until INTEGER,error TEXT,provider_id TEXT);
+CREATE UNIQUE INDEX IF NOT EXISTS outbox_active ON notification_outbox(subscription_id) WHERE state IN ('pending','sending','uncertain');
 `;
 
 export function available() {
@@ -148,6 +150,8 @@ export function list(email) {
       aktif: Boolean(r.aktif),
       sonGonderim: r.sonGonderim,
       initialized: r.sonGorulenler !== null,
+      deliveryPending:!!db.prepare("SELECT id FROM notification_outbox WHERE subscription_id=? AND state IN ('pending','sending','uncertain')").get(r.id),
+      deliveryStatus:db.prepare("SELECT state,error FROM notification_outbox WHERE subscription_id=? AND state IN ('pending','sending','uncertain')").get(r.id) || null,
       sonGorulenler: (() => { try { return JSON.parse(r.sonGorulenler || '[]'); } catch { return []; } })(),
     }));
   } finally {
@@ -159,8 +163,12 @@ export function remove(id, email) {
   const db = open({ create: false });
   if (!db) return false;
   try {
-    const r = db.prepare('DELETE FROM abonelikler WHERE id=? AND email=?').run(id, email);
-    return r.changes > 0;
+    db.exec('BEGIN IMMEDIATE');
+    try {
+      const r=db.prepare('DELETE FROM abonelikler WHERE id=? AND email=?').run(id,email);
+      if(r.changes)db.prepare("UPDATE notification_outbox SET state='cancelled' WHERE subscription_id=? AND state!='sent'").run(id);
+      db.exec('COMMIT');return r.changes>0;
+    }catch(error){db.exec('ROLLBACK');throw error;}
   } finally {
     db.close();
   }
@@ -182,3 +190,48 @@ function recordSeen(id,ihaleIdleri,sent) {
 }
 export const markSent=(id,ids)=>recordSeen(id,ids,true);
 export const markBaseline=(id,ids)=>recordSeen(id,ids,false);
+
+export function reserveDelivery(subscriptionId,payload,tenderIds,{clock=Date.now(),force=false}={}) {
+ const db=open();
+ try {
+  db.exec('BEGIN IMMEDIATE');
+  const subscription=db.prepare('SELECT sonGorulenler FROM abonelikler WHERE id=? AND aktif=1').get(subscriptionId);
+  if(!subscription){db.exec('COMMIT');return null;}
+  let row=db.prepare("SELECT * FROM notification_outbox WHERE subscription_id=? AND state IN ('pending','sending','uncertain')").get(subscriptionId);
+  if(!row){
+   const seen=new Set(JSON.parse(subscription.sonGorulenler || '[]'));
+   if(!payload||!tenderIds.length||(!force&&tenderIds.some(id=>seen.has(id)))){db.exec('COMMIT');return null;}
+   const id=randomUUID();
+   db.prepare("INSERT INTO notification_outbox(id,subscription_id,payload,tender_ids,state) VALUES (?,?,?,?,'pending')").run(id,subscriptionId,JSON.stringify(payload),JSON.stringify(tenderIds));
+   row=db.prepare('SELECT * FROM notification_outbox WHERE id=?').get(id);
+  }
+  if(row.state==='uncertain'||(row.first_attempt!==null&&clock-row.first_attempt>=23*3600000)){
+   db.prepare("UPDATE notification_outbox SET state='uncertain',error=? WHERE id=?").run('Gönderim sonucu belirsiz; sağlayıcı kaydı kontrol edilmeden tekrar gönderilmez.',row.id);
+   db.exec('COMMIT');return {blocked:true};
+  }
+  if(row.state==='sending'&&row.lease_until>clock){db.exec('COMMIT');return null;}
+  const lease=randomUUID();
+  db.prepare("UPDATE notification_outbox SET state='sending',first_attempt=COALESCE(first_attempt,?),lease=?,lease_until=? WHERE id=?").run(clock,lease,clock+120000,row.id);
+  db.exec('COMMIT');return {id:row.id,lease,payload:JSON.parse(row.payload),tenderIds:JSON.parse(row.tender_ids)};
+ }catch(error){try{db.exec('ROLLBACK');}catch{}throw error;}finally{db.close();}
+}
+export function acceptDelivery(delivery,providerId) {
+ if(typeof providerId!=='string'||!providerId)throw new Error('Sağlayıcı gönderim kimliği gerekli.');
+ const db=open();
+ try {
+  db.exec('BEGIN IMMEDIATE');
+  const row=db.prepare("SELECT * FROM notification_outbox WHERE id=? AND lease=? AND state='sending'").get(delivery.id,delivery.lease);
+  if(!row)throw new Error('Gönderim kaydı artık etkin değil.');
+  const subscription=db.prepare('SELECT sonGorulenler FROM abonelikler WHERE id=?').get(row.subscription_id);
+  if(!subscription)throw new Error('Abonelik artık yok.');
+  const seen=[...new Set([...JSON.parse(subscription.sonGorulenler||'[]'),...JSON.parse(row.tender_ids)])];
+  db.prepare('UPDATE abonelikler SET sonGorulenler=?,sonGonderim=? WHERE id=?').run(JSON.stringify(seen),new Date().toISOString(),row.subscription_id);
+  db.prepare("UPDATE notification_outbox SET state='sent',provider_id=?,error=NULL WHERE id=?").run(providerId,row.id);
+  db.exec('COMMIT');
+ }catch(error){try{db.exec('ROLLBACK');}catch{}throw error;}finally{db.close();}
+}
+export function failDelivery(delivery,error) {
+ const db=open();
+ try{return db.prepare("UPDATE notification_outbox SET state='pending',error=? WHERE id=? AND lease=? AND state='sending'").run(String(error?.message || error),delivery.id,delivery.lease).changes>0;}
+ finally{db.close();}
+}