fix(alerts): удерживать неизвестные отправки без автоматического повтора
This commit is contained in:
Denozordec committed
1 parent
46be201d10
commit
b1ad9515fa
9 files changed
+128
-133
No files matched your search
@@ -1,134 +1,40 @@
|
||||
import { and, asc, eq, lte } from "drizzle-orm"
|
||||
import { db } from "../../db/index.js"
|
||||
import { alertOutbox } from "../../db/schema.js"
|
||||
import { appendAlertHistory, sendTelegramAlertMessage } from "../alerts-service.js"
|
||||
import {and,asc,eq,lte} 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 type AlertOutboxPayload = {
|
||||
text: string
|
||||
chatId?: string
|
||||
history: {
|
||||
id: string
|
||||
ruleId?: string | null
|
||||
groupId?: string | null
|
||||
ruleName: string
|
||||
severity: "critical" | "warning" | "info"
|
||||
message: string
|
||||
firedAt: string
|
||||
}
|
||||
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
|
||||
}
|
||||
|
||||
function safeParsePayload(raw: unknown): AlertOutboxPayload | null {
|
||||
if (raw && typeof raw === "object") return raw as AlertOutboxPayload
|
||||
if (typeof raw !== "string") return null
|
||||
try {
|
||||
return JSON.parse(raw) as AlertOutboxPayload
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
export async function enqueueTelegramOutbox(input: {
|
||||
id: string
|
||||
dedupeKey: string
|
||||
payload: AlertOutboxPayload
|
||||
maxRetries?: number
|
||||
}): Promise<boolean> {
|
||||
const existing = (await db
|
||||
.select()
|
||||
.from(alertOutbox)
|
||||
.where(eq(alertOutbox.dedupeKey, input.dedupeKey))
|
||||
.limit(1))[0]
|
||||
if (existing && (existing.status === "pending" || existing.status === "sent")) return false
|
||||
const nowIso = new Date().toISOString()
|
||||
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: nowIso,
|
||||
payloadJson: input.payload,
|
||||
})
|
||||
return true
|
||||
}
|
||||
|
||||
export function dispatchPendingOutbox(limit = 25): Promise<{ sent: number; failed: number; errors: string[] }> {
|
||||
return (async () => {
|
||||
const nowIso = new Date().toISOString()
|
||||
const rows = await db
|
||||
.select()
|
||||
.from(alertOutbox)
|
||||
.where(and(eq(alertOutbox.status, "pending"), lte(alertOutbox.nextAttemptAt, nowIso)))
|
||||
.orderBy(asc(alertOutbox.createdAt))
|
||||
.limit(Math.max(1, limit))
|
||||
|
||||
let sent = 0
|
||||
let failed = 0
|
||||
const errors: string[] = []
|
||||
|
||||
for (const row of rows) {
|
||||
const payload = safeParsePayload(row.payloadJson)
|
||||
if (!payload) {
|
||||
await db.update(alertOutbox)
|
||||
.set({
|
||||
status: "failed",
|
||||
lastError: "invalid payload_json",
|
||||
})
|
||||
.where(eq(alertOutbox.id, row.id))
|
||||
failed += 1
|
||||
continue
|
||||
}
|
||||
const send = await sendTelegramAlertMessage({ text: payload.text, chatId: payload.chatId })
|
||||
if (send.ok) {
|
||||
await db.update(alertOutbox)
|
||||
.set({
|
||||
status: "sent",
|
||||
sentAt: new Date().toISOString(),
|
||||
lastError: null,
|
||||
})
|
||||
.where(eq(alertOutbox.id, row.id))
|
||||
await appendAlertHistory({
|
||||
id: payload.history.id,
|
||||
ruleId: payload.history.ruleId ?? null,
|
||||
groupId: payload.history.groupId ?? null,
|
||||
ruleName: payload.history.ruleName,
|
||||
severity: payload.history.severity,
|
||||
message: payload.history.message,
|
||||
sentOk: true,
|
||||
firedAt: payload.history.firedAt,
|
||||
})
|
||||
sent += 1
|
||||
continue
|
||||
}
|
||||
const nextRetry = row.retryCount + 1
|
||||
const exhausted = nextRetry >= row.maxRetries
|
||||
await db.update(alertOutbox)
|
||||
.set({
|
||||
retryCount: nextRetry,
|
||||
status: exhausted ? "failed" : "pending",
|
||||
nextAttemptAt: new Date(Date.now() + Math.min(300_000, nextRetry * 15_000)).toISOString(),
|
||||
lastError: send.error,
|
||||
})
|
||||
.where(eq(alertOutbox.id, row.id))
|
||||
if (exhausted) {
|
||||
await appendAlertHistory({
|
||||
id: payload.history.id,
|
||||
ruleId: payload.history.ruleId ?? null,
|
||||
groupId: payload.history.groupId ?? null,
|
||||
ruleName: payload.history.ruleName,
|
||||
severity: payload.history.severity,
|
||||
message: payload.history.message,
|
||||
sentOk: false,
|
||||
firedAt: payload.history.firedAt,
|
||||
})
|
||||
failed += 1
|
||||
} else {
|
||||
errors.push(`outbox ${row.id}: ${send.error}`)
|
||||
}
|
||||
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("Журнал результата очереди не подтверждён")
|
||||
}
|
||||
|
||||
return { sent, failed, errors }
|
||||
})()
|
||||
}),
|
||||
})
|
||||
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}
|
||||
}
|
||||
Reference in new issue
Block a user