93 lines
7.4 KiB
TypeScript
93 lines
7.4 KiB
TypeScript
import {executeOutboxRetry} from "../alerts-outbox-retry.js"
|
|
import {executeGuardedApplicationWrite} from "../config-mutation-policy.js"
|
|
import {isDeepStrictEqual} from "node:util"
|
|
import {readOutboxOverview} from "../alerts-outbox-overview.js"
|
|
import {and,asc,desc,eq,inArray,lte,sql} from "drizzle-orm"
|
|
import {db} from "../../db/index.js"
|
|
import {alertOutbox,alertHistory} from "../../db/schema.js"
|
|
import {sendTelegramAlertMessage} from "../alerts-service.js"
|
|
import {executeOutboxDelivery,type AlertOutboxPayload} from "./outbox-execution.js"
|
|
import {assertAlertEngineAccess} from "./execution-access.js"
|
|
export type {AlertOutboxPayload} from "./outbox-execution.js"
|
|
|
|
export async function enqueueTelegramOutbox(input:{id:string;dedupeKey:string;payload:AlertOutboxPayload;maxRetries?:number}):Promise<boolean>{
|
|
await assertAlertEngineAccess()
|
|
const [receipt]=await db.insert(alertOutbox).values({id:input.id,dedupeKey:input.dedupeKey,channel:"telegram",status:"pending",retryCount:0,maxRetries:Math.max(1,Math.floor(input.maxRetries??3)),nextAttemptAt:new Date().toISOString(),payloadJson:input.payload}).onConflictDoNothing({target:alertOutbox.dedupeKey}).returning({id:alertOutbox.id})
|
|
if(!receipt)return false
|
|
if(receipt.id!==input.id)throw Error("Запись сообщения в очередь не подтверждена")
|
|
return true
|
|
}
|
|
|
|
export async function dispatchPendingOutbox(limit=25):Promise<{sent:number;failed:number;held:number;errors:string[]}>{
|
|
const nowIso=new Date().toISOString()
|
|
const rows=await db.select({id:alertOutbox.id}).from(alertOutbox).where(and(eq(alertOutbox.status,"pending"),lte(alertOutbox.nextAttemptAt,nowIso))).orderBy(asc(alertOutbox.createdAt)).limit(Math.max(1,limit))
|
|
let sent=0,failed=0,held=0;const errors:string[]=[]
|
|
for(const row of rows){
|
|
await assertAlertEngineAccess()
|
|
const result=await executeOutboxDelivery(row.id,{
|
|
now:Date.now,
|
|
claim:async id=>{await assertAlertEngineAccess();const [claimed]=await db.update(alertOutbox).set({status:"processing",nextAttemptAt:new Date().toISOString()}).where(and(eq(alertOutbox.id,id),eq(alertOutbox.status,"pending"),lte(alertOutbox.nextAttemptAt,new Date().toISOString()))).returning();if(claimed&&(claimed.id!==id||claimed.status!=="processing"))throw Error("Захват сообщения не подтверждён");return claimed},
|
|
send:async payload=>{await assertAlertEngineAccess();return sendTelegramAlertMessage({text:payload.text,chatId:payload.chatId})},
|
|
settle:(id,value)=>db.transaction(async tx=>{
|
|
const {history,...patch}=value
|
|
await assertAlertEngineAccess()
|
|
const [receipt]=await tx.update(alertOutbox).set(patch).where(and(eq(alertOutbox.id,id),eq(alertOutbox.status,"processing"))).returning({id:alertOutbox.id,status:alertOutbox.status})
|
|
if(!receipt||receipt.id!==id||receipt.status!==value.status)throw Error("Статус очереди не подтверждён")
|
|
if(history){
|
|
await assertAlertEngineAccess()
|
|
const [saved]=await tx.insert(alertHistory).values({...history,ruleId:history.ruleId??null,groupId:history.groupId??null}).returning()
|
|
if(!saved||saved.id!==history.id||saved.sentOk!==history.sentOk||saved.ruleName!==history.ruleName||saved.severity!==history.severity||saved.message!==history.message||saved.ruleId!==(history.ruleId??null)||saved.groupId!==(history.groupId??null)||Date.parse(saved.firedAt)!==Date.parse(history.firedAt))throw Error("Журнал результата очереди не подтверждён")
|
|
}
|
|
}),
|
|
})
|
|
await assertAlertEngineAccess()
|
|
if(result.apiAccepted)sent++
|
|
if(result.kind==="failed")failed++
|
|
if(result.kind==="held")held++
|
|
if(result.error)errors.push("outbox "+row.id+": "+result.error)
|
|
}
|
|
return {sent,failed,held,errors}
|
|
}
|
|
|
|
export async function getAlertsOutboxOverview(){
|
|
return readOutboxOverview<Parameters<Parameters<typeof db.transaction>[0]>[0]>({
|
|
transaction:(work,options)=>db.transaction(work,options),
|
|
counts:async context=>{const result=await context.execute(sql`SELECT status, COUNT(*)::bigint AS count FROM alert_outbox GROUP BY status`);return result.rows},
|
|
items:async context=>{const rows=await context.select().from(alertOutbox).where(inArray(alertOutbox.status,["processing","unknown"])).orderBy(desc(alertOutbox.nextAttemptAt),desc(alertOutbox.id)).limit(20);return rows.map(row=>{const payload=row.payloadJson;const history=typeof payload==="object"&&payload!==null&&"history" in payload?payload.history:null;const ruleName=typeof history==="object"&&history!==null&&"ruleName" in history&&typeof history.ruleName==="string"?history.ruleName:null;return {id:row.id,resolutionVersion:row.resolutionVersion,status:row.status,ruleName,createdAt:row.createdAt,attemptedAt:row.nextAttemptAt}})},
|
|
})
|
|
}
|
|
|
|
export class OutboxRetrySourceUnavailable extends Error {
|
|
constructor(message="Источник записи очереди недоступен или превышает 1 МиБ"){super(message)}
|
|
}
|
|
export type OutboxRetrySource = Record<string, unknown>
|
|
async function readRetrySource(context:Pick<typeof db,"execute">,id:string,lock:boolean):Promise<OutboxRetrySource|null>{
|
|
const result=await context.execute(sql`SELECT CASE WHEN octet_length(to_jsonb(o)::text)<=1048576 THEN to_jsonb(o) END AS source FROM alert_outbox o WHERE id=${id} ${lock?sql`FOR UPDATE OF o`:sql``}`)
|
|
const row=result.rows[0]
|
|
if(!row)return null
|
|
if(!row.source||typeof row.source!=="object"||Array.isArray(row.source))throw new OutboxRetrySourceUnavailable()
|
|
return row.source as OutboxRetrySource
|
|
}
|
|
/** A bounded source for approved adapters; message/recipient fields stay private. */
|
|
export async function readOutboxRetrySource(id:string){return readRetrySource(db,id,false)}
|
|
|
|
export async function retryUnknownOutbox(id:string,input:unknown,guard?:(source:OutboxRetrySource)=>Promise<void>){
|
|
return executeOutboxRetry<Parameters<Parameters<typeof db.transaction>[0]>[0]>(id,input,{
|
|
transaction:work=>db.transaction(work),
|
|
update:async(tx,messageId,expectedVersion)=>{
|
|
let source:OutboxRetrySource|null=null
|
|
if(guard){source=await readRetrySource(tx,messageId,true);if(!source)throw new OutboxRetrySourceUnavailable();await guard(source)}
|
|
return executeGuardedApplicationWrite({kind:"alerts-outbox",id:messageId},"all",async()=>{
|
|
const nextAttemptAt=new Date().toISOString()
|
|
const [receipt]=await tx.update(alertOutbox).set({status:"pending",resolutionVersion:sql`resolution_version + 1`,nextAttemptAt,lastError:null}).where(and(eq(alertOutbox.id,messageId),eq(alertOutbox.status,"unknown"),eq(alertOutbox.channel,"telegram"),eq(alertOutbox.resolutionVersion,expectedVersion))).returning({id:alertOutbox.id,status:alertOutbox.status,resolutionVersion:alertOutbox.resolutionVersion})
|
|
if(source&&receipt){
|
|
const after=await readRetrySource(tx,messageId,true)
|
|
const stable=(value:OutboxRetrySource)=>Object.fromEntries(Object.entries(value).filter(([key])=>!["status","resolution_version","next_attempt_at","last_error"].includes(key)))
|
|
if(!after||after.id!==messageId||after.status!=="pending"||after.resolution_version!==expectedVersion+1||after.last_error!==null||typeof after.next_attempt_at!=="string"||Date.parse(after.next_attempt_at)!==Date.parse(nextAttemptAt)||!isDeepStrictEqual(stable(source),stable(after)))throw new OutboxRetrySourceUnavailable("Не подтверждена неизменность уведомления при постановке в очередь; транзакция отменена")
|
|
}
|
|
return receipt
|
|
})
|
|
},
|
|
})
|
|
}
|