fix: persist weekly brief delivery claims and accepted messages
This commit is contained in:
@@ -1,10 +1,11 @@
|
||||
import { briefWeekKey } from './brief-outbox.mjs';
|
||||
import { splitBriefText } from './brief-content.mjs';
|
||||
|
||||
/** Delivery accounting shared by scheduled and administrator-triggered briefs. */
|
||||
export function createBriefSender({ store, buildBrief, text, html, telegramConfigured, sendTelegram, emailConfigured, sendEmail, warn = console.warn }) {
|
||||
export function createBriefSender({ store, buildBrief, text, html, telegramConfigured, sendTelegram, emailConfigured, sendEmail, outbox, clock = () => new Date(), warn = console.warn }) {
|
||||
return async function sendBrief(tenderGetir) {
|
||||
const tenders = await tenderGetir();
|
||||
const r = { kullanici:0, gonderilen:0, bos:0, hata:0, kapali:0, kanalsiz:0,
|
||||
const r = { kullanici:0, gonderilen:0, bos:0, hata:0, kapali:0, kanalsiz:0, atlanan:0, bekleyen:0,
|
||||
kanallar:{ telegram:{denenen:0,kabul:0,hata:0}, email:{denenen:0,kabul:0,hata:0} } };
|
||||
for (const user of store.listUsers()) {
|
||||
r.kullanici++;
|
||||
@@ -18,28 +19,41 @@ export function createBriefSender({ store, buildBrief, text, html, telegramConfi
|
||||
const telegram = store.getData(user.email, 'telegram');
|
||||
if (telegramConfigured() && telegram?.chatId) {
|
||||
for (const part of splitBriefText(text(brief))) channels.push({
|
||||
payload:{channel:'telegram',chatId:telegram.chatId,text:part},
|
||||
name:'telegram', send:() => sendTelegram(telegram.chatId, part, { parseMode:null }),
|
||||
valid:receipt => receipt === true,
|
||||
});
|
||||
}
|
||||
if (emailConfigured()) channels.push({
|
||||
payload:{channel:'email',to:user.email,subject:`İhaleist — Haftalık Brief (${brief.hafta})`,html:html(brief),text:text(brief)},
|
||||
name:'email', send:() => sendEmail({ to:user.email, subject:`İhaleist — Haftalık Brief (${brief.hafta})`, html:html(brief), text:text(brief) }),
|
||||
valid:receipt => typeof receipt?.id === 'string' && Boolean(receipt.id.trim()),
|
||||
});
|
||||
if (!channels.length) r.kanalsiz++;
|
||||
if (!channels.length) { r.kanalsiz++; continue; }
|
||||
const batch = outbox?.prepare(briefWeekKey(clock()),user.email,channels.map(channel=>channel.payload));
|
||||
if (batch?.changed) { r.bekleyen++; r.hata++; continue; }
|
||||
const failedChannels = new Set();
|
||||
for (const channel of channels) {
|
||||
for (const [position,channel] of channels.entries()) {
|
||||
if (failedChannels.has(channel.name)) continue;
|
||||
// An earlier channel can await the network while the user disables briefs.
|
||||
if (store.getData(user.email, 'brief')?.kapali) { r.kapali++; break; }
|
||||
const claim = outbox?.claim(batch.id,position,clock().getTime());
|
||||
if (claim?.state === 'accepted') { r.atlanan++; continue; }
|
||||
if (claim && claim.state !== 'sending') {
|
||||
r.bekleyen++; failedChannels.add(channel.name);
|
||||
if (claim.state === 'uncertain') failed=true;
|
||||
continue;
|
||||
}
|
||||
const counts = r.kanallar[channel.name];
|
||||
counts.denenen++;
|
||||
try {
|
||||
const receipt = await channel.send();
|
||||
if (!channel.valid(receipt)) throw new Error('Sağlayıcı gönderimi doğrulamadı.');
|
||||
if (claim) outbox.accept(claim,receipt === true ? 'telegram-accepted' : receipt.id);
|
||||
counts.kabul++;
|
||||
accepted = true;
|
||||
} catch (error) {
|
||||
if (claim) outbox.hold(claim,error);
|
||||
counts.hata++;
|
||||
failedChannels.add(channel.name);
|
||||
failed = true;
|
||||
|
||||
@@ -0,0 +1,80 @@
|
||||
import { DatabaseSync } from 'node:sqlite';
|
||||
import { mkdirSync, chmodSync } from 'node:fs';
|
||||
import path from 'node:path';
|
||||
import { randomUUID, createHash } from 'node:crypto';
|
||||
|
||||
const hash = value => createHash('sha256').update(value).digest('hex');
|
||||
export function briefWeekKey(now = new Date()) {
|
||||
const date = new Date(now.getTime() + 3 * 3600000);
|
||||
date.setUTCDate(date.getUTCDate() - (date.getUTCDay()+6)%7);
|
||||
return date.toISOString().slice(0,10);
|
||||
}
|
||||
|
||||
/** A claimed message is never blindly replayed after an interrupted provider call. */
|
||||
export function createBriefOutbox(directory) {
|
||||
function transaction(work) {
|
||||
mkdirSync(directory,{recursive:true,mode:0o700});
|
||||
const file=path.join(directory,'brief.sqlite');
|
||||
const db = new DatabaseSync(file);
|
||||
try {
|
||||
chmodSync(file,0o600);
|
||||
db.exec(`PRAGMA busy_timeout=5000; PRAGMA journal_mode=WAL; PRAGMA synchronous=FULL;
|
||||
CREATE TABLE IF NOT EXISTS brief_batches(id TEXT PRIMARY KEY,week TEXT NOT NULL,email TEXT NOT NULL,manifest TEXT NOT NULL);
|
||||
CREATE TABLE IF NOT EXISTS brief_messages(batch TEXT NOT NULL,position INTEGER NOT NULL,state TEXT NOT NULL,lease TEXT,lease_until INTEGER,receipt TEXT,error TEXT,PRIMARY KEY(batch,position));`);
|
||||
db.exec('BEGIN IMMEDIATE');
|
||||
const result=work(db);
|
||||
db.exec('COMMIT');
|
||||
return result;
|
||||
} catch(error) { try { db.exec('ROLLBACK'); } catch {} throw error; }
|
||||
finally { db.close(); }
|
||||
}
|
||||
return {
|
||||
prepare(week,email,messages) {
|
||||
const id=hash(JSON.stringify([week,email])),manifest=JSON.stringify(messages);
|
||||
return transaction(db=>{
|
||||
const row=db.prepare('SELECT manifest FROM brief_batches WHERE id=?').get(id);
|
||||
if(row) {
|
||||
const changed=row.manifest!==manifest;
|
||||
if(changed) db.prepare("UPDATE brief_messages SET state='uncertain',error=? WHERE batch=? AND state='pending'")
|
||||
.run('Haftalık özet içeriği veya alıcı kanalı değişti; yeniden gönderim öncesi inceleme gerekli.',id);
|
||||
return {id,changed};
|
||||
}
|
||||
db.prepare('INSERT INTO brief_batches VALUES (?,?,?,?)').run(id,week,email,manifest);
|
||||
const insert=db.prepare("INSERT INTO brief_messages(batch,position,state) VALUES (?,?,'pending')");
|
||||
messages.forEach((_,i)=>insert.run(id,i));
|
||||
return {id,changed:false};
|
||||
});
|
||||
},
|
||||
claim(batch,position,clock=Date.now()) {
|
||||
return transaction(db=>{
|
||||
const row=db.prepare('SELECT * FROM brief_messages WHERE batch=? AND position=?').get(batch,position);
|
||||
if(!row)throw new Error('Özet teslimat kaydı bulunamadı.');
|
||||
if(row.state==='accepted')return {state:'accepted'};
|
||||
if(row.state==='uncertain')return {state:'uncertain'};
|
||||
if(row.state==='sending') {
|
||||
if(row.lease_until>clock)return {state:'busy'};
|
||||
db.prepare("UPDATE brief_messages SET state='uncertain',error=? WHERE batch=? AND position=?").run('Kesintiye uğrayan gönderim; sağlayıcı kaydı incelenmeli.',batch,position);
|
||||
return {state:'uncertain'};
|
||||
}
|
||||
const lease=randomUUID();
|
||||
db.prepare("UPDATE brief_messages SET state='sending',lease=?,lease_until=? WHERE batch=? AND position=?").run(lease,clock+120000,batch,position);
|
||||
return {state:'sending',batch,position,lease};
|
||||
});
|
||||
},
|
||||
accept(claim,receipt) {
|
||||
if(typeof receipt!=='string'||!receipt.trim())throw new Error('Gönderim kabul kaydı gerekli.');
|
||||
return transaction(db=>{
|
||||
const result=db.prepare("UPDATE brief_messages SET state='accepted',receipt=?,error=NULL WHERE batch=? AND position=? AND lease=? AND state='sending'")
|
||||
.run(receipt,claim.batch,claim.position,claim.lease);
|
||||
if(result.changes!==1)throw new Error('Özet gönderim kaydı değişti.');
|
||||
});
|
||||
},
|
||||
hold(claim,error) {
|
||||
return transaction(db=>db.prepare("UPDATE brief_messages SET state='uncertain',error=? WHERE batch=? AND position=? AND lease=? AND state='sending'")
|
||||
.run(String(error?.message || error),claim.batch,claim.position,claim.lease).changes===1);
|
||||
},
|
||||
inspect() {
|
||||
return transaction(db=>db.prepare('SELECT b.week,b.email,m.* FROM brief_messages m JOIN brief_batches b ON b.id=m.batch ORDER BY b.week DESC,b.email,m.position').all());
|
||||
},
|
||||
};
|
||||
}
|
||||
+4
-6
@@ -1,8 +1,9 @@
|
||||
// Haftalık brief — Pazartesi 08:00: takipteki ihalelerden bu hafta yapılacaklar,
|
||||
// önceki yedi takvim günündeki takipli sonuç kayıtları.
|
||||
// Telegram (bağlıysa) + e-posta. Son gönderim data/brief-son.json'da tutulur;
|
||||
// saatlik kontrol, haftada bir gönderim (node-cron yok, setInterval).
|
||||
// Telegram (bağlıysa) + e-posta; kalıcı mesaj kabul kayıtları brief.sqlite.
|
||||
// Eski brief-son.json yalnız geçiş haftasında tekrar göndermemek için okunur.
|
||||
import fs from 'node:fs';
|
||||
import { createBriefOutbox } from './brief-outbox.mjs';
|
||||
import { createBriefBuilder } from './brief-content.mjs';
|
||||
import { createBriefSender } from './brief-delivery.mjs';
|
||||
import path from 'node:path';
|
||||
@@ -19,9 +20,6 @@ const DURUM_DOSYASI = path.join(__dirname, 'data', 'brief-son.json');
|
||||
function sonGonderim() {
|
||||
try { return JSON.parse(fs.readFileSync(DURUM_DOSYASI, 'utf8')); } catch { return {}; }
|
||||
}
|
||||
function sonGonderimYaz(d) {
|
||||
try { fs.mkdirSync(path.dirname(DURUM_DOSYASI), { recursive: true }); fs.writeFileSync(DURUM_DOSYASI, JSON.stringify(d)); } catch { /* yok */ }
|
||||
}
|
||||
|
||||
const paraTR = (v, birim) => (v == null ? '' : `${Number(v).toLocaleString('tr-TR')} ${birim || 'TL'}`);
|
||||
|
||||
@@ -62,6 +60,7 @@ export function briefHtml(b) {
|
||||
export const haftalikBriefGonder = createBriefSender({
|
||||
store, buildBrief:briefOlustur, text:briefMetni, html:briefHtml,
|
||||
telegramConfigured, sendTelegram, emailConfigured, sendEmail,
|
||||
outbox:createBriefOutbox(process.env.IHALEIST_USER_DATA_DIR || path.join(__dirname,'data')),
|
||||
});
|
||||
|
||||
export function briefScheduleKey(now = new Date()) {
|
||||
@@ -78,7 +77,6 @@ export function briefZamanlayici(tenderGetir) {
|
||||
if (!anahtar) return;
|
||||
const d = sonGonderim();
|
||||
if (d.son === anahtar) return;
|
||||
sonGonderimYaz({ son: anahtar });
|
||||
const r = await haftalikBriefGonder(tenderGetir).catch((e) => ({ hata: e.message }));
|
||||
console.log('[brief] haftalık brief:', JSON.stringify(r));
|
||||
};
|
||||
|
||||
Reference in New Issue
Block a user