Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 | 63x 3x 2x 2x 2x 2x 2x 2x 3x 3x 1x 3x 2x 2x 10x 10x 5x 5x 3x 3x | import { and, asc, eq, gt, gte, isNotNull, sql } from "drizzle-orm";
import type { DrizzleD1Database } from "drizzle-orm/d1";
import type { NewWaMessage, WaMessage } from "../../db/schema";
import * as schema from "../../db/schema";
import { istDayStart } from "../../lib/ist-time";
import { statusBelow } from "./status-rank";
export class WaMessagesDal {
constructor(private db: DrizzleD1Database<typeof schema>) {}
async findByConversationId(
conversationId: number,
offset = 0,
limit = 50,
): Promise<WaMessage[]> {
return this.db
.select()
.from(schema.waMessages)
.where(eq(schema.waMessages.conversationId, conversationId))
.orderBy(asc(schema.waMessages.dateCreated))
.limit(limit)
.offset(offset);
}
async findById(id: number): Promise<WaMessage | undefined> {
const result = await this.db
.select()
.from(schema.waMessages)
.where(eq(schema.waMessages.id, id))
.limit(1);
return result[0];
}
async findByWamid(wamid: string): Promise<WaMessage | undefined> {
const result = await this.db
.select()
.from(schema.waMessages)
.where(eq(schema.waMessages.wamid, wamid))
.limit(1);
return result[0];
}
async create(data: NewWaMessage): Promise<WaMessage> {
const result = await this.db
.insert(schema.waMessages)
.values(data)
.returning();
return result[0];
}
async updateStatus(
wamid: string,
status: string,
errorCode?: string,
): Promise<void> {
const updateData: Record<string, unknown> = {
status,
dateUpdated: new Date(),
};
if (errorCode) {
updateData.errorCode = errorCode;
}
// Forward-only: a late 'delivered' webhook must not regress 'read'.
await this.db
.update(schema.waMessages)
.set(updateData)
.where(
and(
eq(schema.waMessages.wamid, wamid),
statusBelow(schema.waMessages.status, status),
),
);
}
/** Whether this exact outbound content went to the conversation after `since`. */
async sentSince(
conversationId: number,
content: string,
since: Date,
): Promise<boolean> {
const result = await this.db
.select({ id: schema.waMessages.id })
.from(schema.waMessages)
.where(
and(
eq(schema.waMessages.conversationId, conversationId),
eq(schema.waMessages.direction, "outbound"),
eq(schema.waMessages.content, content),
gt(schema.waMessages.dateCreated, since),
),
)
.limit(1);
return result.length > 0;
}
/**
* Every message, both directions, automated sends included, since 00:00
* IST. Volume, not work done — for that see {@link countAnsweredToday}.
*
* The day starts at IST midnight, not the Worker's. `setHours(0,0,0,0)` in
* a Worker is UTC midnight = 05:30 IST, so it used to credit every message
* between 00:00 and 05:29 IST to the previous day — wrong for five and a
* half hours out of every twenty-four. See src/lib/ist-time.ts.
*/
async countToday(): Promise<number> {
const result = await this.db
.select({ count: sql<number>`count(*)` })
.from(schema.waMessages)
.where(gte(schema.waMessages.dateCreated, istDayStart(new Date())));
return result[0]?.count ?? 0;
}
/**
* Replies a human sent today — the SLA strip's "answered today".
*
* `sent_by` is what separates a human reply from an automated send:
* campaign sends, the notification adapter and the ask-to-resend
* auto-reply all leave it null (see conversation.service.ts). Counting
* outbound alone would make a blast look like a shift's work, which is why
* `countToday()` cannot stand in for this.
*
* Rides wa_messages_created_idx for the date range; `direction` and
* `sent_by` are then row filters.
*/
async countAnsweredToday(): Promise<number> {
const result = await this.db
.select({ count: sql<number>`count(*)` })
.from(schema.waMessages)
.where(
and(
eq(schema.waMessages.direction, "outbound"),
isNotNull(schema.waMessages.sentBy),
gte(schema.waMessages.dateCreated, istDayStart(new Date())),
),
);
return result[0]?.count ?? 0;
}
/**
* Messages pointing at a campaign row that no longer exists.
*
* `wa_messages.campaign_id` is deliberately NOT a foreign key — see the
* column's comment in src/db/schema/whatsapp.ts. This is the assertion
* that stands in for the constraint, and it exists so that "integrity is
* asserted rather than constrained" is a mechanism rather than a promise.
*
* A DIAGNOSTIC: no route calls it, deliberately. It is the check to run
* from a migration or an ops script the day a send path starts writing
* `campaign_id`, and the reason not to delete it as dead code before then.
*/
async countOrphanedCampaignMessages(): Promise<number> {
const result = await this.db
.select({ count: sql<number>`count(*)` })
.from(schema.waMessages)
.where(
and(
isNotNull(schema.waMessages.campaignId),
sql`NOT EXISTS (SELECT 1 FROM ${schema.waCampaigns} WHERE ${schema.waCampaigns.id} = ${schema.waMessages.campaignId})`,
),
);
return result[0]?.count ?? 0;
}
}
|