fix(alerts): показать подтверждённое состояние очереди сообщений
This commit is contained in:
1 parent
b1ad9515fa
commit
89c31289bd
24 files changed
+166
-22
No files matched your search
@@ -1,4 +1,5 @@
|
||||
import {and,asc,eq,lte} from "drizzle-orm"
|
||||
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"
|
||||
@@ -38,3 +39,11 @@ export async function dispatchPendingOutbox(limit=25):Promise<{sent:number;faile
|
||||
}
|
||||
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,status:row.status,ruleName,createdAt:row.createdAt,attemptedAt:row.nextAttemptAt}})},
|
||||
})
|
||||
}
|
||||
@@ -0,0 +1,15 @@
|
||||
import assert from "node:assert/strict"
|
||||
import {test} from "node:test"
|
||||
import {outboxCounts,readOutboxOverview} from "./alerts-outbox-overview.js"
|
||||
test("queue counts keep unknown database states, reject unsafe counts and read one context",async()=>{
|
||||
assert.deepEqual(outboxCounts([]),{total:0,pending:0,processing:0,unknown:0,sent:0,failed:0,unrecognized:0})
|
||||
assert.equal(outboxCounts([{status:"legacy",count:2},{status:"unknown",count:1}]).unrecognized,2)
|
||||
for(const raw of [[{status:"pending",count:-1}],[{status:"pending",count:"1"}],[{status:"sent",count:Number.MAX_SAFE_INTEGER},{status:"failed",count:1}]])assert.throws(()=>outboxCounts(raw))
|
||||
const context={snapshot:1};let reads=0
|
||||
const result=await readOutboxOverview({transaction:async(work,options)=>{assert.deepEqual(options,{isolationLevel:"repeatable read",accessMode:"read only"});return work(context)},counts:async ctx=>{assert.equal(ctx,context);reads++;return [{status:"pending",count:3}]},items:async ctx=>{assert.equal(ctx,context);reads++;return []}})
|
||||
assert.equal(reads,2);assert.equal(result.counts.pending,3)
|
||||
})
|
||||
test("queue incomplete held list or failed read never produces empty success",async()=>{
|
||||
for(const counts of [[{status:"unknown",count:1}],[]])await assert.rejects(readOutboxOverview({transaction:async work=>work({}),counts:async()=>counts,items:async()=>counts.length?[]:[{id:"a",status:"unknown",ruleName:null,createdAt:"2026-10-03T00:00:00Z",attemptedAt:"2026-10-03T00:00:00Z"}]}),/не подтверждено/)
|
||||
await assert.rejects(readOutboxOverview({transaction:async work=>work({}),counts:async()=>{throw Error("read failed")},items:async()=>[]}),/read failed/)
|
||||
})
|
||||
@@ -0,0 +1,16 @@
|
||||
import {z} from "zod"
|
||||
import {alertsOutboxOverviewSchema} from "@mmapp/contracts/alerts"
|
||||
const countsSchema=z.array(z.object({status:z.string(),count:z.number().int().nonnegative().max(Number.MAX_SAFE_INTEGER)}))
|
||||
export function outboxCounts(raw:unknown){
|
||||
const rows=countsSchema.parse(raw),counts={total:0,pending:0,processing:0,unknown:0,sent:0,failed:0,unrecognized:0}
|
||||
for(const row of rows){const key=["pending","processing","unknown","sent","failed"].includes(row.status)?row.status as "pending"|"processing"|"unknown"|"sent"|"failed":"unrecognized";counts[key]+=row.count;counts.total+=row.count}
|
||||
if(!Number.isSafeInteger(counts.total))throw Error("Число записей очереди не подтверждено")
|
||||
return counts
|
||||
}
|
||||
export async function readOutboxOverview<C>(deps:{transaction:(work:(context:C)=>Promise<ReturnType<typeof alertsOutboxOverviewSchema.parse>>,options:{isolationLevel:"repeatable read";accessMode:"read only"})=>Promise<ReturnType<typeof alertsOutboxOverviewSchema.parse>>;counts:(context:C)=>Promise<unknown>;items:(context:C)=>Promise<unknown>}){
|
||||
return deps.transaction(async context=>{
|
||||
const parsed=alertsOutboxOverviewSchema.safeParse({counts:outboxCounts(await deps.counts(context)),items:await deps.items(context),itemsLimit:20})
|
||||
if(!parsed.success)throw Error("Состояние очереди не подтверждено")
|
||||
return parsed.data
|
||||
},{isolationLevel:"repeatable read",accessMode:"read only"})
|
||||
}
|
||||
Reference in new issue
Block a user