Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions .changeset/messaging-shared-unique-violation-predicate.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
---
"@objectstack/service-messaging": patch
---

Migrate the inbox read-receipt race fallback onto the shared `isUniqueViolationError` predicate from `@objectstack/types` (#6542), deleting the last hand-written `isUniqueViolation()` copy that #6250 inventoried. One behaviour change, in the direction the call site wants: the shared predicate follows a bounded step down the `cause` chain, so a unique-constraint conflict wrapped by a pool or query-builder layer now triggers the `flipToRead()` convergence instead of being rethrown as a failed mark-read.
3 changes: 2 additions & 1 deletion packages/services/service-messaging/package.json
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,8 @@
"dependencies": {
"@objectstack/core": "workspace:*",
"@objectstack/platform-objects": "workspace:*",
"@objectstack/spec": "workspace:*"
"@objectstack/spec": "workspace:*",
"@objectstack/types": "workspace:*"
},
"devDependencies": {
"@objectstack/metadata-core": "workspace:*",
Expand Down
74 changes: 74 additions & 0 deletions packages/services/service-messaging/src/messaging-service.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -568,6 +568,80 @@ describe('MessagingService — inbox read API (ADR-0030)', () => {
expect(receipts[0]).toMatchObject({ id: 'r_concurrent', state: 'read' });
});

it('markRead converges when the conflict arrives wrapped in a cause chain (#6542)', async () => {
// Pool and query-builder layers re-throw with the driver error attached
// as `cause`. The retired local isUniqueViolation() read only the
// top-level code/message, judged this wrapper "not a conflict", and
// rethrew — markRead logged the failure and reported readCount 0 with
// the receipt stuck at `delivered`. The shared isUniqueViolationError
// (@objectstack/types, #6250) follows the bounded cause chain, so the
// race fallback now converges exactly as it does for a bare driver
// error. This is the ONE behaviour change of the migration, in the
// direction the call site wants: a wrapped conflict is still a conflict.
const engine = inboxEngine({
inbox: [{ id: 'm1', user_id: 'u1', notification_id: 'n1', title: 'A', created_at: '1' }],
});
const realInsert = engine.insert.bind(engine);
let raced = false;
engine.insert = async (object: string, row: any) => {
if (object === 'sys_notification_receipt' && !raced) {
raced = true;
engine.store.sys_notification_receipt.push({
id: 'r_concurrent', notification_id: 'n1', user_id: 'u1', channel: 'inbox', state: 'delivered',
});
// The wrapper's own text deliberately matches neither the codes
// nor the message substrings the predicate knows — only the
// attached driver error carries the conflict signal.
const wrapper = new Error('receipt insert failed (pool retry exhausted)');
(wrapper as Error & { cause?: unknown }).cause = {
code: '23505',
message: 'duplicate key value violates unique constraint "sys_notification_receipt_notification_id_user_id_channel_key"',
};
throw wrapper;
}
return realInsert(object, row);
};
const svc = new MessagingService({ logger, getData: () => engine });

const res = await svc.markRead('u1', ['n1']);
expect(res).toEqual({ success: true, readCount: 1 });
const receipts = engine.store.sys_notification_receipt;
expect(receipts).toHaveLength(1); // converged, not duplicated
expect(receipts[0]).toMatchObject({ id: 'r_concurrent', state: 'read' });
});

it.each([
['postgres', { code: '23505', message: 'duplicate key value violates unique constraint "sys_notification_receipt_uniq"' }],
['mysql', { code: 'ER_DUP_ENTRY', errno: 1062, message: "Duplicate entry 'n1-u1-inbox' for key 'sys_notification_receipt_uniq'" }],
['sqlite', { message: 'UNIQUE constraint failed: sys_notification_receipt.notification_id, sys_notification_receipt.user_id, sys_notification_receipt.channel' }],
])('markRead race fallback is unchanged for a plain (unwrapped) %s conflict (#6542)', async (_dialect, shape) => {
// Pin that the migration onto the shared predicate did not move any
// verdict the old local copy already gave: a bare three-dialect driver
// error still converges identically.
const engine = inboxEngine({
inbox: [{ id: 'm1', user_id: 'u1', notification_id: 'n1', title: 'A', created_at: '1' }],
});
const realInsert = engine.insert.bind(engine);
let raced = false;
engine.insert = async (object: string, row: any) => {
if (object === 'sys_notification_receipt' && !raced) {
raced = true;
engine.store.sys_notification_receipt.push({
id: 'r_concurrent', notification_id: 'n1', user_id: 'u1', channel: 'inbox', state: 'delivered',
});
throw Object.assign(new Error(shape.message), shape);
}
return realInsert(object, row);
};
const svc = new MessagingService({ logger, getData: () => engine });

const res = await svc.markRead('u1', ['n1']);
expect(res).toEqual({ success: true, readCount: 1 });
const receipts = engine.store.sys_notification_receipt;
expect(receipts).toHaveLength(1);
expect(receipts[0]).toMatchObject({ id: 'r_concurrent', state: 'read' });
});

it('markAllRead flips every unread message and leaves already-read ones', async () => {
const engine = inboxEngine({
inbox: [
Expand Down
21 changes: 2 additions & 19 deletions packages/services/service-messaging/src/messaging-service.ts
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
// Copyright (c) 2025 ObjectStack. Licensed under the Apache-2.0 license.

import type { IDataEngine } from '@objectstack/spec/contracts';
import { isUniqueViolationError } from '@objectstack/types';
import type {
MessagingChannel,
MessagingChannelContext,
Expand All @@ -18,24 +19,6 @@ export const NOTIFICATION_EVENT_OBJECT = 'sys_notification';
/** Receipt states that count as "read" for the inbox unread badge (ADR-0030). */
const READ_RECEIPT_STATES = new Set(['read', 'clicked', 'dismissed']);

/**
* Whether a driver error is a unique/primary-key constraint violation. Spans the
* SQL drivers we ship: SQLite (`UNIQUE constraint failed`), Postgres (`23505` /
* `duplicate key`), and MySQL (`ER_DUP_ENTRY` / `Duplicate entry`). Used to turn
* a lost check-then-act race on a unique index into a fallback update.
*/
function isUniqueViolation(err: unknown): boolean {
const e = err as { code?: string | number; message?: string } | undefined;
if (!e) return false;
if (e.code === '23505' || e.code === 'ER_DUP_ENTRY' || e.code === 'SQLITE_CONSTRAINT_UNIQUE') return true;
const msg = String(e.message ?? '').toLowerCase();
return (
msg.includes('unique constraint failed') ||
msg.includes('duplicate key') ||
msg.includes('duplicate entry')
);
}

/**
* One row of the inbox list REST response — the `Notification` shape in the API
* spec (`NotificationSchema`). `id` is the notification's event id
Expand Down Expand Up @@ -569,7 +552,7 @@ export class MessagingService {
});
return 1;
} catch (err) {
if (isUniqueViolation(err) && (await flipToRead())) return 1;
if (isUniqueViolationError(err) && (await flipToRead())) return 1;
throw err;
}
}
Expand Down
3 changes: 3 additions & 0 deletions pnpm-lock.yaml

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading