58 lines
4.9 KiB
TypeScript
58 lines
4.9 KiB
TypeScript
import {executeOutboxRetry} from "../alerts-outbox-retry.js"
|
|
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"
|
|
export type {AlertOutboxPayload} from "./outbox-execution.js"
|
|
|
|
export async function enqueueTelegramOutbox(input:{id:string;dedupeKey:string;payload:AlertOutboxPayload;maxRetries?:number}):Promise<boolean>{
|
|
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){
|
|
const result=await executeOutboxDelivery(row.id,{
|
|
now:Date.now,
|
|
claim:async id=>{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:payload=>sendTelegramAlertMessage({text:payload.text,chatId:payload.chatId}),
|
|
settle:(id,value)=>db.transaction(async tx=>{
|
|
const {history,...patch}=value
|
|
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){
|
|
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("Журнал результата очереди не подтверждён")
|
|
}
|
|
}),
|
|
})
|
|
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 async function retryUnknownOutbox(id:string,input:unknown){
|
|
return executeOutboxRetry<Parameters<Parameters<typeof db.transaction>[0]>[0]>(id,input,{
|
|
transaction:work=>db.transaction(work),
|
|
update:async(tx,messageId,expectedVersion)=>{const [receipt]=await tx.update(alertOutbox).set({status:"pending",resolutionVersion:sql`resolution_version + 1`,nextAttemptAt:new Date().toISOString(),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});return receipt},
|
|
})
|
|
}
|