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
6 changes: 6 additions & 0 deletions src/db/sqlite/connection.ts
Original file line number Diff line number Diff line change
Expand Up @@ -253,6 +253,7 @@ function createSchema(d: DatabaseSync): void {
server_user_id TEXT NOT NULL,
install_id TEXT NOT NULL,
capability TEXT NOT NULL,
muted TEXT NOT NULL DEFAULT '[]',
created_at TEXT NOT NULL,
updated_at TEXT NOT NULL,
PRIMARY KEY (server_user_id, install_id)
Expand Down Expand Up @@ -710,6 +711,11 @@ function hasColumn(d: DatabaseSync, table: string, column: string): boolean {
}

function runMigrations(d: DatabaseSync): void {
// Conversations muted on that phone (GRYT-1689). Only dev servers ever had the table without it.
if (!hasColumn(d, "push_devices", "muted")) {
d.exec("ALTER TABLE push_devices ADD COLUMN muted TEXT NOT NULL DEFAULT '[]'");
}

const cols = d.prepare("PRAGMA table_info(users)").all() as { name: string }[];
const colNames = new Set(cols.map((c) => c.name));

Expand Down
2 changes: 1 addition & 1 deletion src/db/sqlite/pushDevices.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@ describe("push devices", () => {
it("keeps one row per install and replaces its capability", () => {
savePushDevice("u1", "install-1", cap("a"));
savePushDevice("u1", "install-1", cap("b"));
assert.deepEqual(listPushDevices("u1"), [{ installId: "install-1", capability: cap("b") }]);
assert.deepEqual(listPushDevices("u1"), [{ installId: "install-1", capability: cap("b"), muted: new Set() }]);
removePushDevice("u1", "install-1");
assert.deepEqual(listPushDevices("u1"), []);
});
Expand Down
46 changes: 36 additions & 10 deletions src/db/sqlite/pushDevices.ts
Original file line number Diff line number Diff line change
Expand Up @@ -4,29 +4,55 @@ import { getSqliteDb, toIso } from "./connection";
export interface PushDevice {
installId: string;
capability: string;
/** Conversations muted on that phone, which never reach the relay (GRYT-1689). */
muted: ReadonlySet<string>;
}

/** More than this and the oldest goes. Nobody has ten phones; a reinstall loop might. */
export const MAX_PUSH_DEVICES_PER_USER = 10;

export function savePushDevice(serverUserId: string, installId: string, capability: string): void {
/** The phone checks in on every connect. One silent this long is probably in a drawer. */
export const PUSH_DEVICE_STALE_DAYS = 30;

const DAY_MS = 24 * 60 * 60 * 1000;

function parseMuted(raw: string): ReadonlySet<string> {
try {
const value = JSON.parse(raw) as unknown;
return new Set(Array.isArray(value) ? value.filter((v): v is string => typeof v === "string") : []);
} catch {
return new Set();
}
}

export function savePushDevice(
serverUserId: string,
installId: string,
capability: string,
muted: readonly string[] = [],
now = new Date(),
): void {
const db = getSqliteDb();
const now = toIso(new Date());
const at = toIso(now);
db.prepare(
`INSERT INTO push_devices (server_user_id, install_id, capability, created_at, updated_at) VALUES (?, ?, ?, ?, ?)
ON CONFLICT(server_user_id, install_id) DO UPDATE SET capability = excluded.capability, updated_at = excluded.updated_at`,
).run(serverUserId, installId, capability, now, now);
`INSERT INTO push_devices (server_user_id, install_id, capability, muted, created_at, updated_at) VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(server_user_id, install_id) DO UPDATE SET capability = excluded.capability, muted = excluded.muted, updated_at = excluded.updated_at`,
).run(serverUserId, installId, capability, JSON.stringify(muted), at, at);
db.prepare(
`DELETE FROM push_devices WHERE server_user_id = ? AND install_id NOT IN (
SELECT install_id FROM push_devices WHERE server_user_id = ? ORDER BY updated_at DESC, install_id LIMIT ?)`,
).run(serverUserId, serverUserId, MAX_PUSH_DEVICES_PER_USER);
}

export function listPushDevices(serverUserId: string): PushDevice[] {
const rows = getSqliteDb()
.prepare(`SELECT install_id, capability FROM push_devices WHERE server_user_id = ?`)
.all(serverUserId) as { install_id: string; capability: string }[];
return rows.map((r) => ({ installId: r.install_id, capability: r.capability }));
/** Only phones that checked in lately. The rest are dropped here, one account at a time. */
export function listPushDevices(serverUserId: string, now = new Date()): PushDevice[] {
const db = getSqliteDb();
const cutoff = toIso(new Date(now.getTime() - PUSH_DEVICE_STALE_DAYS * DAY_MS));
db.prepare(`DELETE FROM push_devices WHERE server_user_id = ? AND updated_at < ?`).run(serverUserId, cutoff);
const rows = db
.prepare(`SELECT install_id, capability, muted FROM push_devices WHERE server_user_id = ?`)
.all(serverUserId) as { install_id: string; capability: string; muted: string }[];
return rows.map((r) => ({ installId: r.install_id, capability: r.capability, muted: parseMuted(r.muted) }));
}

export function removePushDevice(serverUserId: string, installId: string): void {
Expand Down
2 changes: 2 additions & 0 deletions src/db/sqlite/users.ts
Original file line number Diff line number Diff line change
Expand Up @@ -288,6 +288,8 @@ export async function getAllAvatarFileIds(): Promise<Set<string>> {
export async function setUserInactive(serverUserId: string): Promise<void> {
const db = getSqliteDb();
db.prepare(`UPDATE users SET is_active = 0 WHERE server_user_id = ?`).run(serverUserId);
// Leaving, a kick and a ban all land here. A phone that comes back registers again.
db.prepare(`DELETE FROM push_devices WHERE server_user_id = ?`).run(serverUserId);
}

/** `carried` hands the guest's row to the account; `merged` folds it into the row
Expand Down
10 changes: 5 additions & 5 deletions src/services/push.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -57,14 +57,14 @@ describe("who is at a screen", () => {

describe("pushing", () => {
it("sends the capability in the header and only the kind in the body", async () => {
const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A }] });
const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A, muted: new Set<string>() }] });
h.pusher.notify({}, ["u1"], "dm", "conv");
await flush();
assert.deepEqual(h.calls, [{ url: "https://push.test/v1/push", auth: `Bearer ${CAP_A}`, body: JSON.stringify({ kind: "dm" }) }]);
});

it("skips somebody who is at a screen, and wakes them once they put the phone away", async () => {
const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A }] });
const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A, muted: new Set<string>() }] });
h.pusher.notify({ s: client("u1") }, ["u1"], "mention", "conv");
await flush();
assert.equal(h.calls.length, 0);
Expand All @@ -74,7 +74,7 @@ describe("pushing", () => {
});

it("buzzes once per phone per conversation in fifteen seconds", async () => {
const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A }, { installId: "tablet-12", capability: CAP_B }] });
const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A, muted: new Set<string>() }, { installId: "tablet-12", capability: CAP_B, muted: new Set<string>() }] });
h.pusher.notify({}, ["u1"], "dm", "conv");
h.pusher.notify({}, ["u1", "u1"], "dm", "conv");
h.pusher.notify({}, ["u1"], "dm", "other");
Expand All @@ -86,7 +86,7 @@ describe("pushing", () => {

it("forgets a capability the relay calls gone or unknown", async () => {
for (const status of [404, 410]) {
const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A }] }, status);
const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A, muted: new Set<string>() }] }, status);
h.pusher.notify({}, ["u1"], "dm", "conv");
await flush();
await flush();
Expand All @@ -95,7 +95,7 @@ describe("pushing", () => {
});

it("keeps it when the relay is only busy", async () => {
const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A }] }, 502);
const h = harness({ u1: [{ installId: "phone-1234", capability: CAP_A, muted: new Set<string>() }] }, 502);
h.pusher.notify({}, ["u1"], "dm", "conv");
await flush();
await flush();
Expand Down
7 changes: 7 additions & 0 deletions src/services/push.ts
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,7 @@ export function createPusher(deps: PusherDeps) {
continue;
}
for (const device of devices) {
if (device.muted.has(conversationId)) continue;
const key = `${device.capability}:${conversationId}`;
if (now - (lastSent.get(key) ?? 0) < QUIET_MS) continue;
lastSent.set(key, now);
Expand Down Expand Up @@ -118,6 +119,12 @@ export function pushNotify(clientsInfo: Clients, serverUserIds: Iterable<string>
shared.notify(clientsInfo, serverUserIds, kind, conversationId);
}

/** Tests only: forget the throttle and read the relay setting again. */
export function resetPushState(): void {
shared = null;
relay = undefined;
}

export function pushEnabled(): boolean {
return configuredRelay() !== null;
}
18 changes: 16 additions & 2 deletions src/socket/handlers/push.ts
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,15 @@ type Reply = { ok: true } | { ok: false; error: string };
type Ack = (reply: Reply) => void;

const INSTALL_ID = /^[A-Za-z0-9_-]{8,64}$/;
const MAX_MUTED = 1000;

/** Conversation ids the phone muted. Null when it is not a short list of short strings. */
function mutedFrom(raw: unknown): string[] | null {
if (raw === undefined) return [];
if (!Array.isArray(raw) || raw.length > MAX_MUTED) return null;
if (!raw.every((id) => typeof id === "string" && id.length > 0 && id.length <= 128)) return null;
return [...new Set(raw as string[])];
}
const RL_REGISTER: RateLimitRule = { limit: 20, windowMs: 60_000, scorePerAction: 1, maxScore: 12, scoreDecayMs: 3_000 };

export function registerPushHandlers(ctx: HandlerContext): EventHandlerMap {
Expand All @@ -25,7 +34,10 @@ export function registerPushHandlers(ctx: HandlerContext): EventHandlerMap {
}

return {
"push:register": async (payload: { accessToken: string; installId?: unknown; capability?: unknown }, ack: Ack) => {
"push:register": async (
payload: { accessToken: string; installId?: unknown; capability?: unknown; muted?: unknown },
ack: Ack,
) => {
ack = typeof ack === "function" ? ack : () => {};
try {
if (!pushEnabled()) return ack({ ok: false, error: "push_off" });
Expand All @@ -34,9 +46,11 @@ export function registerPushHandlers(ctx: HandlerContext): EventHandlerMap {
|| typeof payload.capability !== "string" || !CAPABILITY_SHAPE.test(payload.capability)) {
return ack({ ok: false, error: "invalid_payload" });
}
const muted = mutedFrom(payload.muted);
if (!muted) return ack({ ok: false, error: "invalid_payload" });
const auth = await requireAuth(socket, payload);
if (!auth) return ack({ ok: false, error: "unauthorized" });
savePushDevice(auth.tokenPayload.serverUserId, payload.installId, payload.capability);
savePushDevice(auth.tokenPayload.serverUserId, payload.installId, payload.capability, muted);
ack({ ok: true });
} catch (err) {
consola.error("push:register failed", err);
Expand Down
Loading
Loading