pepepper/fluxer
1
0 Fork 0
mirror of https://github.com/fluxerapp/fluxer synced 2026-10-17 21:15:32 +09:00
Code Issues Packages Projects Releases Wiki Activity

feat(message): remove the unfinished scheduled messages feature (#2233)

Browse Source
This commit is contained in:
Hampus authored and GitHub committed 2026-08-31 13:32:55 +02:00
1 parent eeea391b63
commit 3c8466d714
125 files changed
+17 -3730
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

fluxer_api/src/api/Tables.ts
-7
View File
@@ -305,8 +305,6 @@ import {
type RelationshipRow,
SAVED_MESSAGE_COLUMNS,
type SavedMessageRow,
SCHEDULED_MESSAGE_COLUMNS,
type ScheduledMessageRow,
USER_BY_EMAIL_COLUMNS,
USER_BY_LAST_ACTIVE_IP_COLUMNS,
USER_BY_LAST_ACTIVE_IP_TRUST_KEY_COLUMNS,
@@ -684,11 +682,6 @@ export const SavedMessages = defineTable<SavedMessageRow, 'user_id' | 'message_i
columns: SAVED_MESSAGE_COLUMNS,
primaryKey: ['user_id', 'message_id'],
});
export const ScheduledMessages = defineTable<ScheduledMessageRow, 'user_id' | 'scheduled_message_id'>({
name: 'scheduled_messages',
columns: SCHEDULED_MESSAGE_COLUMNS,
primaryKey: ['user_id', 'scheduled_message_id'],
});
export const PushSubscriptions = defineTable<PushSubscriptionRow, 'user_id' | 'subscription_id'>({
name: 'push_subscriptions',
columns: PUSH_SUBSCRIPTION_COLUMNS,
fluxer_api/src/api/ZodErrorMap.ts
-2
View File
@@ -66,8 +66,6 @@ function fluxerZodErrorMap(issue: FluxerZodErrorMapIssue): FluxerZodErrorMapResu
const origin = 'origin' in issue ? String(issue.origin) : undefined;
if (origin === 'string') {
errorCode = ValidationErrorCodes.CONTENT_EXCEEDS_MAX_LENGTH;
} else if (origin === 'date') {
errorCode = ValidationErrorCodes.SCHEDULED_TIME_MUST_BE_FUTURE;
} else {
errorCode = ValidationErrorCodes.INVALID_FORMAT;
}
fluxer_api/src/api/channel/controllers/ScheduledMessageController.ts
-55
View File
@@ -1,55 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import {ChannelIdParam} from '@fluxer/schema/src/domains/common/CommonParamSchemas';
import {
ScheduledMessageRequestSchema,
ScheduledMessageResponseSchema,
} from '@fluxer/schema/src/domains/message/ScheduledMessageSchemas';
import {createChannelID} from '../../BrandedTypes';
import {DefaultUserOnly, LoginRequired} from '../../middleware/AuthMiddleware';
import {RateLimitMiddleware} from '../../middleware/RateLimitMiddleware';
import {OpenAPI} from '../../middleware/ResponseTypeMiddleware';
import {RateLimitConfigs} from '../../RateLimitConfig';
import type {HonoApp} from '../../types/HonoEnv';
import {Validator} from '../../Validator';
import {parseScheduledMessageInput} from './ScheduledMessageParsing';
export function ScheduledMessageController(app: HonoApp) {
app.post(
'/channels/:channel_id/messages/schedule',
RateLimitMiddleware(RateLimitConfigs.CHANNEL_MESSAGE_CREATE),
LoginRequired,
DefaultUserOnly,
Validator('param', ChannelIdParam),
OpenAPI({
operationId: 'schedule_message',
summary: 'Schedule a message to send later',
description:
'Schedules a message to be sent at a specified time. Only available for regular user accounts. Requires permission to send messages in the target channel. Message is sent automatically at the scheduled time. Returns the scheduled message object with delivery time.',
responseSchema: ScheduledMessageResponseSchema,
requestSchema: ScheduledMessageRequestSchema,
requestFormSchema: ScheduledMessageRequestSchema,
statusCode: 201,
security: ['bearerToken', 'sessionToken'],
tags: 'Channels',
}),
async (ctx) => {
const user = ctx.get('user');
const channelId = createChannelID(ctx.req.valid('param').channel_id);
const scheduledMessageService = ctx.get('scheduledMessageService');
const {message, scheduledLocalAt, timezone} = await parseScheduledMessageInput({
ctx,
user,
channelId,
});
const scheduledMessage = await scheduledMessageService.createScheduledMessage({
user,
channelId,
data: message,
scheduledLocalAt,
timezone,
});
return ctx.json(scheduledMessage.toResponse(), 201);
},
);
}
fluxer_api/src/api/channel/controllers/ScheduledMessageParsing.ts
-78
View File
@@ -1,78 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import {ValidationErrorCodes} from '@fluxer/constants/src/ValidationErrorCodes';
import {InputValidationError} from '@fluxer/errors/src/domains/core/InputValidationError';
import {MessageRequestSchema} from '@fluxer/schema/src/domains/message/MessageRequestSchemas';
import {ScheduledMessageRequestSchema} from '@fluxer/schema/src/domains/message/ScheduledMessageSchemas';
import type {Context} from 'hono';
import {ms} from 'itty-time';
import type {ChannelID} from '../../BrandedTypes';
import type {User} from '../../models/User';
import type {HonoEnv} from '../../types/HonoEnv';
import {parseJsonPreservingLargeIntegers} from '../../utils/LosslessJsonParser';
import type {MessageRequest} from '../MessageTypes';
import {normalizeMessageRequestPayload} from '../services/message/MessageRequestCompatibility';
import {parseMultipartMessageData} from '../services/message/MessageRequestParser';
type ScheduledMessageSchemaType = ScheduledMessageRequestSchema;
function extractScheduleFields(data: ScheduledMessageSchemaType): {
scheduled_local_at: string;
timezone: string;
message: MessageRequest;
} {
const {scheduled_local_at, timezone, ...messageData} = data;
return {
scheduled_local_at,
timezone,
message: messageData as MessageRequest,
};
}
export async function parseScheduledMessageInput({
ctx,
user,
channelId,
}: {
ctx: Context<HonoEnv>;
user: User;
channelId: ChannelID;
}): Promise<{
message: MessageRequest;
scheduledLocalAt: string;
timezone: string;
}> {
const contentType = ctx.req.header('content-type') ?? '';
const isMultipart = contentType.includes('multipart/form-data');
if (isMultipart) {
let parsedPayload: unknown = null;
const message = (await parseMultipartMessageData(ctx, user, channelId, MessageRequestSchema, {
uploadExpiresAt: new Date(Date.now() + ms('32 days')),
onPayloadParsed(payload) {
parsedPayload = payload;
},
})) as MessageRequest;
if (!parsedPayload) {
throw InputValidationError.fromCode('scheduled_message', ValidationErrorCodes.FAILED_TO_PARSE_MULTIPART_PAYLOAD);
}
const validation = ScheduledMessageRequestSchema.safeParse(normalizeMessageRequestPayload(parsedPayload));
if (!validation.success) {
throw InputValidationError.fromCode('scheduled_message', ValidationErrorCodes.INVALID_SCHEDULED_MESSAGE_PAYLOAD);
}
const {scheduled_local_at, timezone} = extractScheduleFields(validation.data);
return {message, scheduledLocalAt: scheduled_local_at, timezone};
}
let body: unknown;
try {
const raw = await ctx.req.text();
body = raw.trim().length === 0 ? {} : parseJsonPreservingLargeIntegers(raw);
} catch {
throw InputValidationError.fromCode('scheduled_message', ValidationErrorCodes.INVALID_SCHEDULED_MESSAGE_PAYLOAD);
}
const validation = ScheduledMessageRequestSchema.safeParse(normalizeMessageRequestPayload(body));
if (!validation.success) {
throw InputValidationError.fromCode('scheduled_message', ValidationErrorCodes.INVALID_SCHEDULED_MESSAGE_PAYLOAD);
}
const {scheduled_local_at, timezone, message} = extractScheduleFields(validation.data);
return {message, scheduledLocalAt: scheduled_local_at, timezone};
}
fluxer_api/src/api/channel/controllers/index.ts
-2
View File
@@ -5,14 +5,12 @@ import {CallController} from './CallController';
import {ChannelController} from './ChannelController';
import {MessageController} from './MessageController';
import {MessageInteractionController} from './MessageInteractionController';
import {ScheduledMessageController} from './ScheduledMessageController';
import {StreamController} from './StreamController';
export function registerChannelControllers(app: HonoApp) {
ChannelController(app);
MessageInteractionController(app);
MessageController(app);
ScheduledMessageController(app);
CallController(app);
StreamController(app);
}
fluxer_api/src/api/channel/services/ScheduledMessageService.ts
-185
View File
@@ -1,185 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import {UserFlags} from '@fluxer/constants/src/UserConstants';
import {ValidationErrorCodes} from '@fluxer/constants/src/ValidationErrorCodes';
import {FeatureTemporarilyDisabledError} from '@fluxer/errors/src/domains/core/FeatureTemporarilyDisabledError';
import {InputValidationError} from '@fluxer/errors/src/domains/core/InputValidationError';
import type {IWorkerService} from '@pkgs/worker/src/contracts/IWorkerService';
import type {WorkerJobPayload} from '@pkgs/worker/src/contracts/WorkerTypes';
import {ms, seconds} from 'itty-time';
import {DateTime, IANAZone} from 'luxon';
import type {ChannelID, MessageID, UserID} from '../../BrandedTypes';
import {createMessageID} from '../../BrandedTypes';
import type {ISnowflakeService} from '../../infrastructure/ISnowflakeService';
import type {ScheduledMessagePayload} from '../../models/ScheduledMessage';
import {ScheduledMessage} from '../../models/ScheduledMessage';
import type {User} from '../../models/User';
import type {ScheduledMessageRepository} from '../../user/repositories/ScheduledMessageRepository';
import type {WorkerTaskName} from '../../worker/WorkerLaneConfig';
import type {MessageRequest} from '../MessageTypes';
import type {ChannelService} from './ChannelService';
export const SCHEDULED_MESSAGE_TTL_SECONDS = seconds('31 days');
const DEFAULT_TIMEZONE = 'UTC';
const WORKER_TASK_NAME = 'sendScheduledMessage';
interface ScheduleParams {
user: User;
channelId: ChannelID;
data: MessageRequest;
scheduledLocalAt: string;
timezone?: string;
}
interface UpdateScheduleParams extends ScheduleParams {
scheduledMessageId: MessageID;
existing?: ScheduledMessage;
}
interface SendScheduledMessageWorkerPayload extends WorkerJobPayload {
userId: string;
scheduledMessageId: string;
expectedScheduledAt: string;
}
export class ScheduledMessageService {
constructor(
private readonly channelService: ChannelService,
private readonly scheduledMessageRepository: ScheduledMessageRepository,
private readonly workerService: IWorkerService<WorkerTaskName>,
private readonly snowflakeService: ISnowflakeService,
) {}
async listScheduledMessages(userId: UserID): Promise<Array<ScheduledMessage>> {
return await this.scheduledMessageRepository.listScheduledMessages(userId);
}
async getScheduledMessage(userId: UserID, scheduledMessageId: MessageID): Promise<ScheduledMessage | null> {
return await this.scheduledMessageRepository.getScheduledMessage(userId, scheduledMessageId);
}
async createScheduledMessage(params: ScheduleParams): Promise<ScheduledMessage> {
return await this.upsertScheduledMessage({
...params,
scheduledMessageId: await createSnowflake(this.snowflakeService),
});
}
async updateScheduledMessage(params: UpdateScheduleParams): Promise<ScheduledMessage> {
return await this.upsertScheduledMessage({
...params,
existing:
params.existing ?? (await this.getScheduledMessage(params.user.id, params.scheduledMessageId)) ?? undefined,
});
}
async cancelScheduledMessage(userId: UserID, scheduledMessageId: MessageID): Promise<void> {
await this.scheduledMessageRepository.deleteScheduledMessage(userId, scheduledMessageId);
}
async markInvalid(userId: UserID, scheduledMessageId: MessageID, reason: string): Promise<void> {
await this.scheduledMessageRepository.markInvalid(
userId,
scheduledMessageId,
reason,
SCHEDULED_MESSAGE_TTL_SECONDS,
);
}
private async upsertScheduledMessage(params: UpdateScheduleParams): Promise<ScheduledMessage> {
const {user, channelId, data, scheduledLocalAt, timezone} = params;
if ((user.flags & UserFlags.STAFF) === 0n) {
throw new FeatureTemporarilyDisabledError();
}
await this.channelService.messages.send.validateMessageCanBeSent({
user,
channelId,
data,
});
const scheduledAt = this.resolveScheduledAt(scheduledLocalAt, timezone);
const payload = toScheduledPayload(data);
const existing = params.existing;
const message = new ScheduledMessage({
userId: user.id,
id: params.scheduledMessageId,
channelId,
scheduledAt,
scheduledLocalAt,
timezone: timezone ?? DEFAULT_TIMEZONE,
payload,
status: 'pending',
statusReason: null,
createdAt: existing?.createdAt,
invalidatedAt: null,
});
await this.scheduledMessageRepository.upsertScheduledMessage(message, SCHEDULED_MESSAGE_TTL_SECONDS);
await this.scheduleWorker(message);
return message;
}
private resolveScheduledAt(local: string, timezone?: string): Date {
const zone = timezone?.trim() || DEFAULT_TIMEZONE;
if (!IANAZone.isValidZone(zone)) {
throw InputValidationError.fromCode('timezone', ValidationErrorCodes.INVALID_TIMEZONE_IDENTIFIER);
}
const dt = DateTime.fromISO(local, {zone});
if (!dt.isValid) {
throw InputValidationError.fromCode(
'scheduled_local_at',
ValidationErrorCodes.INVALID_DATETIME_FOR_SCHEDULED_SEND,
);
}
const scheduledAt = dt.toJSDate();
const now = Date.now();
const diffMs = scheduledAt.getTime() - now;
if (diffMs <= 0) {
throw InputValidationError.fromCode('scheduled_local_at', ValidationErrorCodes.SCHEDULED_TIME_MUST_BE_FUTURE);
}
if (diffMs > ms('30 days')) {
throw InputValidationError.fromCode('scheduled_local_at', ValidationErrorCodes.SCHEDULED_MESSAGES_MAX_30_DAYS);
}
return scheduledAt;
}
private async scheduleWorker(message: ScheduledMessage): Promise<void> {
const payload: SendScheduledMessageWorkerPayload = {
userId: message.userId.toString(),
scheduledMessageId: message.id.toString(),
expectedScheduledAt: message.scheduledAt.toISOString(),
};
await this.workerService.addJob(WORKER_TASK_NAME, payload, {runAt: message.scheduledAt});
}
}
async function createSnowflake(snowflakeService: ISnowflakeService): Promise<MessageID> {
return createMessageID(await snowflakeService.generate());
}
function toScheduledPayload(data: MessageRequest): ScheduledMessagePayload {
return {
content: data.content ?? null,
embeds: data.embeds,
attachments: data.attachments,
message_reference: data.message_reference
? {
message_id: data.message_reference.message_id.toString(),
channel_id: data.message_reference.channel_id?.toString(),
guild_id: data.message_reference.guild_id?.toString(),
type: data.message_reference.type,
}
: undefined,
allowed_mentions: data.allowed_mentions
? {
parse: data.allowed_mentions.parse,
users: data.allowed_mentions.users?.map((id) => id.toString()),
roles: data.allowed_mentions.roles?.map((id) => id.toString()),
replied_user: data.allowed_mentions.replied_user,
}
: undefined,
flags: data.flags,
nonce: data.nonce,
favorite_meme_id: data.favorite_meme_id?.toString(),
sticker_ids: data.sticker_ids?.map((id) => id.toString()),
tts: data.tts,
};
}
fluxer_api/src/api/channel/tests/ScheduledMessageTestUtils.ts
-177
View File
@@ -1,177 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import type {MessageResponse} from '@fluxer/schema/src/domains/message/MessageResponseSchemas';
import type {ScheduledMessageResponseSchema} from '@fluxer/schema/src/domains/message/ScheduledMessageSchemas';
import type {z} from 'zod';
import {ensureSessionStarted} from '../../message/tests/MessageTestUtils';
import type {ApiTestHarness} from '../../test/ApiTestHarness';
import {createBuilder, createBuilderWithoutAuth} from '../../test/TestRequestBuilder';
type ScheduledMessageResponse = z.infer<typeof ScheduledMessageResponseSchema>;
export async function createGuildChannel(
harness: ApiTestHarness,
token: string,
guildId: string,
name: string,
): Promise<{
id: string;
}> {
const {response, json} = await createBuilder<{
id: string;
}>(harness, token)
.post(`/guilds/${guildId}/channels`)
.body({name, type: 0})
.executeWithResponse();
if (response.status !== 200 && response.status !== 201) {
throw new Error(`Failed to create guild channel: ${response.status}`);
}
return json as {
id: string;
};
}
export async function grantStaffAccess(harness: ApiTestHarness, userId: string): Promise<void> {
await createBuilderWithoutAuth(harness).patch(`/test/users/${userId}/flags`).body({flags: 1}).execute();
}
export async function scheduleMessage(
harness: ApiTestHarness,
channelId: string,
token: string,
content: string,
scheduledAt?: Date,
): Promise<ScheduledMessageResponse> {
const scheduledTime = scheduledAt ?? new Date(Date.now() + 60 * 1000);
await ensureSessionStarted(harness, token);
const {json} = await createBuilder<ScheduledMessageResponse>(harness, token)
.post(`/channels/${channelId}/messages/schedule`)
.body({
content,
scheduled_local_at: scheduledTime.toISOString(),
timezone: 'UTC',
})
.expect(201)
.executeWithResponse();
return json as ScheduledMessageResponse;
}
export async function updateScheduledMessage(
harness: ApiTestHarness,
scheduledMessageId: string,
token: string,
updates: {
content?: string;
scheduled_local_at?: string;
timezone?: string;
},
): Promise<ScheduledMessageResponse> {
const {json} = await createBuilder<ScheduledMessageResponse>(harness, token)
.patch(`/users/@me/scheduled-messages/${scheduledMessageId}`)
.body(updates)
.expect(200)
.executeWithResponse();
return json as ScheduledMessageResponse;
}
export async function getScheduledMessages(
harness: ApiTestHarness,
token: string,
): Promise<Array<ScheduledMessageResponse>> {
const {json} = await createBuilder<Array<ScheduledMessageResponse>>(harness, token)
.get('/users/@me/scheduled-messages')
.expect(200)
.executeWithResponse();
return json;
}
export async function getScheduledMessage(
harness: ApiTestHarness,
scheduledMessageId: string,
token: string,
expectedStatus: 200 | 404 = 200,
): Promise<ScheduledMessageResponse | null> {
const {response, json} = await createBuilder<ScheduledMessageResponse>(harness, token)
.get(`/users/@me/scheduled-messages/${scheduledMessageId}`)
.expect(expectedStatus)
.executeWithResponse();
if (response.status === 404) {
return null;
}
return json as ScheduledMessageResponse;
}
export async function cancelScheduledMessage(
harness: ApiTestHarness,
scheduledMessageId: string,
token: string,
): Promise<void> {
await createBuilder<void>(harness, token)
.delete(`/users/@me/scheduled-messages/${scheduledMessageId}`)
.expect(204)
.execute();
}
export async function triggerScheduledMessageWorker(
harness: ApiTestHarness,
userId: string,
scheduledMessageId: string,
): Promise<void> {
await createBuilder<void>(harness, '')
.post(`/test/worker/send-scheduled-message/${userId}/${scheduledMessageId}`)
.expect(200)
.execute();
}
export async function getChannelMessages(
harness: ApiTestHarness,
channelId: string,
token: string,
limit = 20,
): Promise<Array<MessageResponse>> {
const {json} = await createBuilder<Array<MessageResponse>>(harness, token)
.get(`/channels/${channelId}/messages?limit=${limit}`)
.expect(200)
.executeWithResponse();
return json;
}
export function messageFromAuthorContains(
messages: Array<MessageResponse>,
authorId: string,
content: string,
): boolean {
return messages.some((msg) => msg.author.id === authorId && msg.content === content);
}
export async function createChannelInvite(
harness: ApiTestHarness,
token: string,
channelId: string,
): Promise<{
code: string;
}> {
const {json} = await createBuilder<{
code: string;
}>(harness, token)
.post(`/channels/${channelId}/invites`)
.body({
max_age: 86400,
})
.expect(200)
.executeWithResponse();
return json;
}
export async function joinGuild(harness: ApiTestHarness, token: string, inviteCode: string): Promise<void> {
await createBuilder<void>(harness, token).post(`/invites/${inviteCode}`).body({}).expect(200).execute();
}
export async function removeGuildMember(
harness: ApiTestHarness,
token: string,
guildId: string,
userId: string,
): Promise<void> {
await createBuilder<void>(harness, token).delete(`/guilds/${guildId}/members/${userId}`).expect(204).execute();
}
fluxer_api/src/api/channel/tests/ScheduledMessageTraitGated.test.ts
-38
View File
@@ -1,38 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import {beforeAll, beforeEach, describe, expect, it} from 'vitest';
import {createTestAccount} from '../../auth/tests/AuthTestUtils';
import {createGuild} from '../../guild/tests/GuildTestUtils';
import {ensureSessionStarted} from '../../message/tests/MessageTestUtils';
import {type ApiTestHarness, createApiTestHarness} from '../../test/ApiTestHarness';
import {createBuilder} from '../../test/TestRequestBuilder';
import {createGuildChannel, grantStaffAccess, scheduleMessage} from './ScheduledMessageTestUtils';
describe('Scheduled message staff gating', () => {
let harness: ApiTestHarness;
beforeAll(async () => {
harness = await createApiTestHarness();
});
beforeEach(async () => {
await harness.reset();
});
it('rejects scheduling message before staff flag granted', async () => {
const owner = await createTestAccount(harness);
const guild = await createGuild(harness, owner.token, 'scheduled-flag');
const channel = await createGuildChannel(harness, owner.token, guild.id, 'scheduled-channel');
await ensureSessionStarted(harness, owner.token);
await createBuilder(harness, owner.token)
.post(`/channels/${channel.id}/messages/schedule`)
.body({
content: 'trying to schedule',
scheduled_local_at: new Date(Date.now() + 60 * 1000).toISOString(),
timezone: 'UTC',
})
.expect(403)
.execute();
await grantStaffAccess(harness, owner.userId);
const scheduled = await scheduleMessage(harness, channel.id, owner.token, 'enabled now');
expect(scheduled.id).toBeDefined();
expect(scheduled.id).not.toBe('');
});
});
fluxer_api/src/api/channel/tests/ScheduledMessageValidation.test.ts
-131
View File
@@ -1,131 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import {beforeAll, beforeEach, describe, expect, it} from 'vitest';
import {createTestAccount} from '../../auth/tests/AuthTestUtils';
import {createGuild} from '../../guild/tests/GuildTestUtils';
import {ensureSessionStarted} from '../../message/tests/MessageTestUtils';
import {type ApiTestHarness, createApiTestHarness} from '../../test/ApiTestHarness';
import {generateFutureTimestamp, generatePastTimestamp, HTTP_STATUS, TEST_LIMITS} from '../../test/TestConstants';
import {createBuilder} from '../../test/TestRequestBuilder';
import {createGuildChannel, grantStaffAccess, scheduleMessage} from './ScheduledMessageTestUtils';
describe('Scheduled message validation', () => {
let harness: ApiTestHarness;
beforeAll(async () => {
harness = await createApiTestHarness();
});
beforeEach(async () => {
await harness.reset();
});
it('rejects scheduling message with past time', async () => {
const owner = await createTestAccount(harness);
const guild = await createGuild(harness, owner.token, 'sched-validation-past');
await grantStaffAccess(harness, owner.userId);
const channel = await createGuildChannel(harness, owner.token, guild.id, 'test');
await ensureSessionStarted(harness, owner.token);
const pastTime = generatePastTimestamp(1);
const {json} = await createBuilder<{
errors: Array<{
path: string;
message: string;
code: string;
}>;
}>(harness, owner.token)
.post(`/channels/${channel.id}/messages/schedule`)
.body({
content: 'should fail - past time',
scheduled_local_at: pastTime,
timezone: 'UTC',
})
.expect(HTTP_STATUS.BAD_REQUEST)
.executeWithResponse();
const errorJson = json as {
errors: Array<{
path: string;
message: string;
code: string;
}>;
};
const hasScheduledAtError = errorJson.errors.some((e) => e.path === 'scheduled_local_at');
expect(hasScheduledAtError).toBe(true);
});
it('rejects scheduling message exceeding 30 days', async () => {
const owner = await createTestAccount(harness);
const guild = await createGuild(harness, owner.token, 'sched-validation-30day');
await grantStaffAccess(harness, owner.userId);
const channel = await createGuildChannel(harness, owner.token, guild.id, 'test');
await ensureSessionStarted(harness, owner.token);
const futureTime = new Date(Date.now() + TEST_LIMITS.SCHEDULED_MESSAGE_MAX_DELAY_MS).toISOString();
const {json} = await createBuilder<{
errors: Array<{
path: string;
message: string;
code: string;
}>;
}>(harness, owner.token)
.post(`/channels/${channel.id}/messages/schedule`)
.body({
content: 'should fail - exceeds 30 days',
scheduled_local_at: futureTime,
timezone: 'UTC',
})
.expect(HTTP_STATUS.BAD_REQUEST)
.executeWithResponse();
const errorJson = json as {
errors: Array<{
path: string;
message: string;
code: string;
}>;
};
const hasScheduledAtError = errorJson.errors.some((e) => e.path === 'scheduled_local_at');
expect(hasScheduledAtError).toBe(true);
});
it('rejects scheduling message with invalid timezone', async () => {
const owner = await createTestAccount(harness);
const guild = await createGuild(harness, owner.token, 'sched-validation-tz');
await grantStaffAccess(harness, owner.userId);
const channel = await createGuildChannel(harness, owner.token, guild.id, 'test');
await ensureSessionStarted(harness, owner.token);
const futureTime = generateFutureTimestamp(5);
const {json} = await createBuilder<{
errors: Array<{
path: string;
message: string;
code: string;
}>;
}>(harness, owner.token)
.post(`/channels/${channel.id}/messages/schedule`)
.body({
content: 'should fail - invalid timezone',
scheduled_local_at: futureTime,
timezone: 'Invalid/NotATimezone',
})
.expect(HTTP_STATUS.BAD_REQUEST)
.executeWithResponse();
const errorJson = json as {
errors: Array<{
path: string;
message: string;
code: string;
}>;
};
const hasTimezoneError = errorJson.errors.some((e) => e.path === 'timezone');
expect(hasTimezoneError).toBe(true);
});
it('accepts scheduling message at 30 day boundary', async () => {
const owner = await createTestAccount(harness);
const guild = await createGuild(harness, owner.token, 'sched-validation-boundary');
await grantStaffAccess(harness, owner.userId);
const channel = await createGuildChannel(harness, owner.token, guild.id, 'test');
const futureTime = new Date(Date.now() + 29 * 24 * 60 * 60 * 1000 + 23 * 60 * 60 * 1000).toISOString();
const scheduled = await scheduleMessage(
harness,
channel.id,
owner.token,
'should succeed - within 30 days',
new Date(futureTime),
);
expect(scheduled.status).toBe('pending');
});
});
fluxer_api/src/api/channel/tests/ScheduledMessageWorkerLifecycle.test.ts
-91
View File
@@ -1,91 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import {beforeAll, beforeEach, describe, expect, it} from 'vitest';
import {createTestAccount} from '../../auth/tests/AuthTestUtils';
import {createGuild} from '../../guild/tests/GuildTestUtils';
import {type ApiTestHarness, createApiTestHarness} from '../../test/ApiTestHarness';
import {
createChannelInvite,
createGuildChannel,
getChannelMessages,
getScheduledMessage,
grantStaffAccess,
joinGuild,
messageFromAuthorContains,
removeGuildMember,
scheduleMessage,
triggerScheduledMessageWorker,
updateScheduledMessage,
} from './ScheduledMessageTestUtils';
describe('Scheduled message worker lifecycle', () => {
let harness: ApiTestHarness;
beforeAll(async () => {
harness = await createApiTestHarness();
});
beforeEach(async () => {
await harness.reset();
});
it('delivers scheduled message when permissions remain', async () => {
const owner = await createTestAccount(harness);
const guild = await createGuild(harness, owner.token, 'scheduled-messages');
await grantStaffAccess(harness, owner.userId);
const channel = await createGuildChannel(harness, owner.token, guild.id, 'scheduled');
const content = 'scheduled message goes through';
const scheduled = await scheduleMessage(harness, channel.id, owner.token, content);
await triggerScheduledMessageWorker(harness, owner.userId, scheduled.id);
const fetched = await getScheduledMessage(harness, scheduled.id, owner.token, 404);
expect(fetched).toBeNull();
const messages = await getChannelMessages(harness, channel.id, owner.token);
expect(messageFromAuthorContains(messages, owner.userId, content)).toBe(true);
});
it('reschedules pending message before worker execution', async () => {
const owner = await createTestAccount(harness);
const guild = await createGuild(harness, owner.token, 'scheduled-messages');
await grantStaffAccess(harness, owner.userId);
const channel = await createGuildChannel(harness, owner.token, guild.id, 'scheduled');
const content = 'scheduled message initial content';
const scheduled = await scheduleMessage(harness, channel.id, owner.token, content);
const oldScheduledAt = new Date(scheduled.scheduled_at);
const updatedContent = 'scheduled message updated content';
const newLocalTime = new Date(Date.now() + 5 * 60 * 1000);
const newLocalStr = newLocalTime.toISOString();
const updated = await updateScheduledMessage(harness, scheduled.id, owner.token, {
content: updatedContent,
scheduled_local_at: newLocalStr,
timezone: 'America/Los_Angeles',
});
expect(updated.status).toBe('pending');
expect(updated.scheduled_local_at).toBe(newLocalStr);
expect(updated.timezone).toBe('America/Los_Angeles');
const updatedScheduledAt = new Date(updated.scheduled_at);
expect(updatedScheduledAt.getTime()).toBeGreaterThan(oldScheduledAt.getTime());
await triggerScheduledMessageWorker(harness, owner.userId, updated.id);
const respMessages = await getChannelMessages(harness, channel.id, owner.token);
expect(messageFromAuthorContains(respMessages, owner.userId, updatedContent)).toBe(true);
expect(messageFromAuthorContains(respMessages, owner.userId, content)).toBe(false);
const fetched = await getScheduledMessage(harness, updated.id, owner.token, 404);
expect(fetched).toBeNull();
});
it('marks scheduled message invalid when access lost', async () => {
const owner = await createTestAccount(harness);
const member = await createTestAccount(harness);
const guild = await createGuild(harness, owner.token, 'scheduled-messages');
await grantStaffAccess(harness, member.userId);
const channel = await createGuildChannel(harness, owner.token, guild.id, 'scheduled');
const invite = await createChannelInvite(harness, owner.token, channel.id);
await joinGuild(harness, member.token, invite.code);
const content = 'scheduled message invalidation';
const scheduled = await scheduleMessage(harness, channel.id, member.token, content);
await removeGuildMember(harness, owner.token, guild.id, member.userId);
await triggerScheduledMessageWorker(harness, member.userId, scheduled.id);
const fetched = await getScheduledMessage(harness, scheduled.id, member.token);
expect(fetched).not.toBeNull();
expect(fetched!.status).toBe('invalid');
expect(fetched!.status_reason).not.toBeNull();
expect(fetched!.status_reason).not.toBe('');
const messages = await getChannelMessages(harness, channel.id, owner.token);
const hasContent = messages.some((msg) => msg.content === content);
expect(hasContent).toBe(false);
});
});
fluxer_api/src/api/channel/tests/ScheduledMessagesListInvalidEntry.test.ts
-45
View File
@@ -1,45 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import {beforeAll, beforeEach, describe, expect, it} from 'vitest';
import {createTestAccount} from '../../auth/tests/AuthTestUtils';
import {createGuild} from '../../guild/tests/GuildTestUtils';
import {type ApiTestHarness, createApiTestHarness} from '../../test/ApiTestHarness';
import {
createChannelInvite,
createGuildChannel,
getScheduledMessages,
grantStaffAccess,
joinGuild,
removeGuildMember,
scheduleMessage,
triggerScheduledMessageWorker,
} from './ScheduledMessageTestUtils';
describe('Scheduled messages list invalid entry', () => {
let harness: ApiTestHarness;
beforeAll(async () => {
harness = await createApiTestHarness();
});
beforeEach(async () => {
await harness.reset();
});
it('shows invalid scheduled message in list', async () => {
const owner = await createTestAccount(harness);
const member = await createTestAccount(harness);
const guild = await createGuild(harness, owner.token, 'scheduled-invalid');
await grantStaffAccess(harness, member.userId);
const channel = await createGuildChannel(harness, owner.token, guild.id, 'scheduled-invalid');
const invite = await createChannelInvite(harness, owner.token, channel.id);
await joinGuild(harness, member.token, invite.code);
const content = 'invalid scheduled';
const scheduled = await scheduleMessage(harness, channel.id, member.token, content);
await removeGuildMember(harness, owner.token, guild.id, member.userId);
await triggerScheduledMessageWorker(harness, member.userId, scheduled.id);
const list = await getScheduledMessages(harness, member.token);
const entry = list.find((e) => e.id === scheduled.id);
expect(entry).toBeDefined();
expect(entry!.status).toBe('invalid');
expect(entry!.status_reason).not.toBeNull();
expect(entry!.status_reason).not.toBe('');
});
});
fluxer_api/src/api/channel/tests/ScheduledMessagesListLifecycle.test.ts
-38
View File
@@ -1,38 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import {beforeAll, beforeEach, describe, expect, it} from 'vitest';
import {createTestAccount} from '../../auth/tests/AuthTestUtils';
import {createGuild} from '../../guild/tests/GuildTestUtils';
import {type ApiTestHarness, createApiTestHarness} from '../../test/ApiTestHarness';
import {
cancelScheduledMessage,
createGuildChannel,
getScheduledMessages,
grantStaffAccess,
scheduleMessage,
} from './ScheduledMessageTestUtils';
describe('Scheduled messages list lifecycle', () => {
let harness: ApiTestHarness;
beforeAll(async () => {
harness = await createApiTestHarness();
});
beforeEach(async () => {
await harness.reset();
});
it('lists scheduled messages and removes after cancel', async () => {
const owner = await createTestAccount(harness);
const guild = await createGuild(harness, owner.token, 'scheduled-list');
await grantStaffAccess(harness, owner.userId);
const channel = await createGuildChannel(harness, owner.token, guild.id, 'scheduled-list');
const content = 'list scheduled';
const scheduled = await scheduleMessage(harness, channel.id, owner.token, content);
const list = await getScheduledMessages(harness, owner.token);
const found = list.some((entry) => entry.id === scheduled.id);
expect(found).toBe(true);
await cancelScheduledMessage(harness, scheduled.id, owner.token);
const listAfterCancel = await getScheduledMessages(harness, owner.token);
const foundAfterCancel = listAfterCancel.some((entry) => entry.id === scheduled.id);
expect(foundAfterCancel).toBe(false);
});
});
fluxer_api/src/api/database/types/UserTypes.ts
-28
View File
@@ -343,34 +343,6 @@ export const SAVED_MESSAGE_COLUMNS = [
'saved_at',
] as const satisfies ReadonlyArray<keyof SavedMessageRow>;
export interface ScheduledMessageRow {
user_id: UserID;
scheduled_message_id: MessageID;
channel_id: ChannelID;
payload: string;
scheduled_at: Date;
scheduled_local_at: string;
timezone: string;
status: string;
status_reason: string | null;
created_at: Date;
invalidated_at: Date | null;
}
export const SCHEDULED_MESSAGE_COLUMNS = [
'user_id',
'scheduled_message_id',
'channel_id',
'payload',
'scheduled_at',
'scheduled_local_at',
'timezone',
'status',
'status_reason',
'created_at',
'invalidated_at',
] as const satisfies ReadonlyArray<keyof ScheduledMessageRow>;
export interface FavoriteMemeRow {
user_id: UserID;
meme_id: MemeID;
fluxer_api/src/api/infrastructure/KVScheduledJobQueueService.ts
-118
View File
@@ -1,118 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import {createHash} from 'node:crypto';
import type {IKVProvider} from '@pkgs/kv_client/src/IKVProvider';
import type {WorkerJobPayload} from '@pkgs/worker/src/contracts/WorkerTypes';
import {Logger} from '../Logger';
export interface ParkedScheduledJob {
jobIdentity: string;
taskType: string;
payload: WorkerJobPayload;
runAtMs: number;
ledgerJobId: string | null;
}
export interface ReadyScheduledJob {
job: ParkedScheduledJob;
member: string;
}
const QUEUE_KEY = 'scheduled_job_queue';
const SECONDARY_KEY_PREFIX = 'scheduled_job_queue:';
export function buildScheduledJobIdentity(
taskType: string,
payload: WorkerJobPayload,
runAtMs: number,
ledgerJobId: string | null,
): string {
if (ledgerJobId !== null) {
return ledgerJobId;
}
return createHash('sha256')
.update(`${taskType}|${runAtMs}|${JSON.stringify(payload)}`)
.digest('hex');
}
export class KVScheduledJobQueueService {
constructor(private readonly kvClient: IKVProvider) {}
private getSecondaryKey(jobIdentity: string): string {
return `${SECONDARY_KEY_PREFIX}${jobIdentity}`;
}
private serializeQueueItem(job: ParkedScheduledJob): string {
return JSON.stringify(job);
}
private deserializeQueueItem(value: string): ParkedScheduledJob {
const parsed: unknown = JSON.parse(value);
if (typeof parsed !== 'object' || parsed === null) {
throw new Error('parked scheduled job must be a JSON object');
}
const {jobIdentity, taskType, payload, runAtMs, ledgerJobId} = parsed as Record<string, unknown>;
if (typeof jobIdentity !== 'string' || typeof taskType !== 'string' || typeof runAtMs !== 'number') {
throw new Error('parked scheduled job is missing required fields');
}
if (typeof payload !== 'object' || payload === null || Array.isArray(payload)) {
throw new Error('parked scheduled job payload must be a JSON object');
}
return {
jobIdentity,
taskType,
payload: payload as WorkerJobPayload,
runAtMs,
ledgerJobId: typeof ledgerJobId === 'string' ? ledgerJobId : null,
};
}
async parkJob(job: ParkedScheduledJob, releaseAt: Date): Promise<void> {
try {
const secondaryKey = this.getSecondaryKey(job.jobIdentity);
const value = this.serializeQueueItem(job);
await this.kvClient.removeBulkDeletion(QUEUE_KEY, secondaryKey);
await this.kvClient.scheduleBulkDeletion(QUEUE_KEY, secondaryKey, releaseAt.getTime(), value);
Logger.debug({jobIdentity: job.jobIdentity, taskType: job.taskType, releaseAt}, 'Parked scheduled job');
} catch (error) {
Logger.error({error, jobIdentity: job.jobIdentity, taskType: job.taskType}, 'Failed to park scheduled job');
throw error;
}
}
async claimJob(jobIdentity: string, member = ''): Promise<boolean> {
try {
return await this.kvClient.removeBulkDeletion(QUEUE_KEY, this.getSecondaryKey(jobIdentity), member);
} catch (error) {
Logger.error({error, jobIdentity}, 'Failed to claim parked scheduled job');
throw error;
}
}
async getReadyJobs(nowMs: number, limit: number): Promise<Array<ReadyScheduledJob>> {
try {
const results = await this.kvClient.zrangebyscore(QUEUE_KEY, '-inf', nowMs, 'LIMIT', 0, limit);
const jobs: Array<ReadyScheduledJob> = [];
for (const result of results) {
try {
jobs.push({job: this.deserializeQueueItem(result), member: result});
} catch (error) {
Logger.error({error, result}, 'Failed to parse parked scheduled job entry');
}
}
return jobs;
} catch (error) {
Logger.error({error, nowMs, limit}, 'Failed to fetch ready parked scheduled jobs');
throw error;
}
}
async getQueueSize(): Promise<number> {
try {
return await this.kvClient.zcard(QUEUE_KEY);
} catch (error) {
Logger.error({error}, 'Failed to get parked scheduled job queue size');
throw error;
}
}
}
fluxer_api/src/api/middleware/ServiceMiddleware.ts
-13
View File
@@ -21,7 +21,6 @@ import {ChannelRepository} from '../channel/ChannelRepository';
import {ChannelRequestService} from '../channel/services/ChannelRequestService';
import {MessageRequestService} from '../channel/services/message/MessageRequestService';
import {createMessageResponseDataService} from '../channel/services/message/MessageResponseDataService';
import {ScheduledMessageService} from '../channel/services/ScheduledMessageService';
import {StreamService} from '../channel/services/StreamService';
import {ConnectionRequestService} from '../connection/ConnectionRequestService';
import {ConnectionService} from '../connection/ConnectionService';
@@ -148,7 +147,6 @@ import {
getReadStateRequestService,
getReadStateService,
getReportRepository,
getScheduledMessageRepository,
getStorageService,
getStreamPreviewService,
getSweegoWebhookService,
@@ -384,7 +382,6 @@ class RequestServices implements RequestScopedServices {
private cachedUserChannelRequestService: UserChannelRequestService | undefined;
private cachedUserContentRequestService: UserContentRequestService | undefined;
private cachedUserRelationshipRequestService: UserRelationshipRequestService | undefined;
private cachedScheduledMessageService: ScheduledMessageService | undefined;
private cachedWebhookService: WebhookService | undefined;
private cachedWebhookRequestService: WebhookRequestService | undefined;
@@ -1022,16 +1019,6 @@ class RequestServices implements RequestScopedServices {
return this.cachedUserRelationshipRequestService;
}
get scheduledMessageService(): ScheduledMessageService {
this.cachedScheduledMessageService ??= new ScheduledMessageService(
this.channelService,
getScheduledMessageRepository(),
getWorkerService(),
getSnowflakeService(),
);
return this.cachedScheduledMessageService;
}
get webhookService(): WebhookService {
this.cachedWebhookService ??= new WebhookService(
getWebhookRepository(),
fluxer_api/src/api/middleware/ServiceSingletons.ts
-15
View File
@@ -60,7 +60,6 @@ import type {IUnfurlerService} from '../infrastructure/IUnfurlerService';
import {KVAccountDeletionQueueService} from '../infrastructure/KVAccountDeletionQueueService';
import {KVActivityTracker} from '../infrastructure/KVActivityTracker';
import {KVBulkMessageDeletionQueueService} from '../infrastructure/KVBulkMessageDeletionQueueService';
import {KVScheduledJobQueueService} from '../infrastructure/KVScheduledJobQueueService';
import {NatsUnfurlerService} from '../infrastructure/NatsUnfurlerService';
import {PremiumStateReconciliationQueueService} from '../infrastructure/PremiumStateReconciliationQueueService';
import {createDownloadsStorageService, createStorageService} from '../infrastructure/StorageServiceFactory';
@@ -88,7 +87,6 @@ import {EntranceSoundRepository} from '../user/entrance_sound/EntranceSoundRepos
import {EntranceSoundService} from '../user/entrance_sound/EntranceSoundService';
import {EmailChangeRepository} from '../user/repositories/auth/EmailChangeRepository';
import {PasswordChangeRepository} from '../user/repositories/auth/PasswordChangeRepository';
import {ScheduledMessageRepository} from '../user/repositories/ScheduledMessageRepository';
import {UserContactChangeLogRepository} from '../user/repositories/UserContactChangeLogRepository';
import {UserRepository} from '../user/repositories/UserRepository';
import {VisionarySlotRepository} from '../user/repositories/VisionarySlotRepository';
@@ -123,7 +121,6 @@ export const getApplicationRepository = singleton(() => new ApplicationRepositor
export const getOAuth2TokenRepository = singleton(() => new OAuth2TokenRepository());
export const getPackRepository = singleton(() => new PackRepository());
export const getGuildDiscoveryRepository = singleton(() => new GuildDiscoveryRepository());
export const getScheduledMessageRepository = singleton(() => new ScheduledMessageRepository());
export const getEmailChangeRepository = singleton(() => new EmailChangeRepository());
export const getPasswordChangeRepository = singleton(() => new PasswordChangeRepository());
const getUserContactChangeLogRepository = singleton(() => new UserContactChangeLogRepository());
@@ -230,18 +227,6 @@ export function getKVBulkMessageDeletionQueue(): KVBulkMessageDeletionQueueServi
return bulkMessageDeletionQueue;
}
let scheduledJobQueueClient: IKVProvider | null = null;
let scheduledJobQueue: KVScheduledJobQueueService | null = null;
export function getKVScheduledJobQueue(): KVScheduledJobQueueService {
const kvClient = getKVClient();
if (!scheduledJobQueue || scheduledJobQueueClient !== kvClient) {
scheduledJobQueue = new KVScheduledJobQueueService(kvClient);
scheduledJobQueueClient = kvClient;
}
return scheduledJobQueue;
}
let premiumStateQueueClient: IKVProvider | null = null;
let premiumStateQueue: PremiumStateReconciliationQueueService | null = null;
fluxer_api/src/api/middleware/tests/LazyServiceProvider.test.ts
-1
View File
@@ -63,7 +63,6 @@ const REQUEST_SERVICE_VARIABLES: ReadonlyArray<keyof HonoEnv['Variables']> = [
'reportRequestService',
'reportService',
'rpcService',
'scheduledMessageService',
'searchService',
'singleCommunityService',
'snowflakeService',
fluxer_api/src/api/models/ScheduledMessage.ts
-220
View File
@@ -1,220 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import type {MessageReferenceTypeValue} from '@fluxer/constants/src/ChannelConstants';
import type {AllowedMentionsRequest} from '@fluxer/schema/src/domains/message/SharedMessageSchemas';
import type {ChannelID, MessageID, UserID} from '../BrandedTypes';
import type {AttachmentRequestData} from '../channel/AttachmentDTOs';
import type {MessageRequest} from '../channel/MessageTypes';
import type {ScheduledMessageRow} from '../database/types/UserTypes';
import {isJsonRecord, parseJsonWithGuard} from '../utils/JsonBoundaryUtils';
type ScheduledMessageStatus = 'pending' | 'invalid';
export interface ScheduledMessageReference {
message_id: string;
channel_id?: string;
guild_id?: string;
type?: MessageReferenceTypeValue;
}
export interface ScheduledAllowedMentions {
parse?: AllowedMentionsRequest['parse'];
users?: Array<string>;
roles?: Array<string>;
replied_user?: boolean;
}
export interface ScheduledMessagePayload {
content?: string | null;
embeds?: MessageRequest['embeds'];
attachments?: Array<AttachmentRequestData>;
message_reference?: ScheduledMessageReference;
allowed_mentions?: ScheduledAllowedMentions;
flags?: number;
nonce?: string;
favorite_meme_id?: string;
sticker_ids?: Array<string>;
tts?: boolean;
}
interface ScheduledMessageResponse {
id: string;
channel_id: string;
scheduled_at: string;
scheduled_local_at: string;
timezone: string;
status: ScheduledMessageStatus;
status_reason: string | null;
payload: ScheduledMessagePayload;
created_at: string;
invalidated_at: string | null;
}
export class ScheduledMessage {
readonly userId: UserID;
readonly id: MessageID;
readonly channelId: ChannelID;
readonly scheduledAt: Date;
readonly scheduledLocalAt: string;
readonly timezone: string;
readonly payload: ScheduledMessagePayload;
readonly status: ScheduledMessageStatus;
readonly statusReason: string | null;
readonly createdAt: Date;
readonly invalidatedAt: Date | null;
constructor(params: {
userId: UserID;
id: MessageID;
channelId: ChannelID;
scheduledAt: Date;
scheduledLocalAt: string;
timezone: string;
payload: ScheduledMessagePayload;
status?: ScheduledMessageStatus;
statusReason?: string | null;
createdAt?: Date;
invalidatedAt?: Date | null;
}) {
this.userId = params.userId;
this.id = params.id;
this.channelId = params.channelId;
this.scheduledAt = params.scheduledAt;
this.scheduledLocalAt = params.scheduledLocalAt;
this.timezone = params.timezone;
this.payload = params.payload;
this.status = params.status ?? 'pending';
this.statusReason = params.statusReason ?? null;
this.createdAt = params.createdAt ?? new Date();
this.invalidatedAt = params.invalidatedAt ?? null;
}
static fromRow(row: ScheduledMessageRow): ScheduledMessage {
return new ScheduledMessage({
userId: row.user_id,
id: row.scheduled_message_id,
channelId: row.channel_id,
scheduledAt: row.scheduled_at,
scheduledLocalAt: row.scheduled_local_at,
timezone: row.timezone,
payload: parsePayload(row.payload),
status: toStatus(row.status),
statusReason: row.status_reason,
createdAt: row.created_at,
invalidatedAt: row.invalidated_at,
});
}
toRow(): ScheduledMessageRow {
return {
user_id: this.userId,
scheduled_message_id: this.id,
channel_id: this.channelId,
payload: JSON.stringify(this.payload),
scheduled_at: this.scheduledAt,
scheduled_local_at: this.scheduledLocalAt,
timezone: this.timezone,
status: this.status,
status_reason: this.statusReason,
created_at: this.createdAt,
invalidated_at: this.invalidatedAt,
};
}
toResponse(): ScheduledMessageResponse {
return {
id: this.id.toString(),
channel_id: this.channelId.toString(),
scheduled_at: this.scheduledAt.toISOString(),
scheduled_local_at: this.scheduledLocalAt,
timezone: this.timezone,
status: this.status,
status_reason: this.statusReason,
payload: this.payload,
created_at: this.createdAt.toISOString(),
invalidated_at: this.invalidatedAt?.toISOString() ?? null,
};
}
parseToMessageRequest(): MessageRequest {
return hydrateMessageRequest(this.payload);
}
}
function parsePayload(payload: string): ScheduledMessagePayload {
return parseJsonWithGuard(payload, isScheduledMessagePayload) ?? {};
}
function isStringArray(value: unknown): value is Array<string> {
return Array.isArray(value) && value.every((entry) => typeof entry === 'string');
}
function isScheduledMessageReference(value: unknown): value is ScheduledMessageReference {
if (!isJsonRecord(value)) return false;
return (
typeof value.message_id === 'string' &&
(value.channel_id === undefined || typeof value.channel_id === 'string') &&
(value.guild_id === undefined || typeof value.guild_id === 'string') &&
(value.type === undefined || typeof value.type === 'number')
);
}
function isScheduledAllowedMentions(value: unknown): value is ScheduledAllowedMentions {
if (!isJsonRecord(value)) return false;
return (
(value.parse === undefined || isStringArray(value.parse)) &&
(value.users === undefined || isStringArray(value.users)) &&
(value.roles === undefined || isStringArray(value.roles)) &&
(value.replied_user === undefined || typeof value.replied_user === 'boolean')
);
}
function isScheduledMessagePayload(value: unknown): value is ScheduledMessagePayload {
if (!isJsonRecord(value)) return false;
return (
(value.content === undefined || value.content === null || typeof value.content === 'string') &&
(value.embeds === undefined || Array.isArray(value.embeds)) &&
(value.attachments === undefined || Array.isArray(value.attachments)) &&
(value.message_reference === undefined || isScheduledMessageReference(value.message_reference)) &&
(value.allowed_mentions === undefined || isScheduledAllowedMentions(value.allowed_mentions)) &&
(value.flags === undefined || typeof value.flags === 'number') &&
(value.nonce === undefined || typeof value.nonce === 'string') &&
(value.favorite_meme_id === undefined || typeof value.favorite_meme_id === 'string') &&
(value.sticker_ids === undefined || isStringArray(value.sticker_ids)) &&
(value.tts === undefined || typeof value.tts === 'boolean')
);
}
function toStatus(value: string): ScheduledMessageStatus {
return value === 'invalid' ? 'invalid' : 'pending';
}
function hydrateMessageRequest(payload: ScheduledMessagePayload): MessageRequest {
const message: MessageRequest = {
content: payload.content,
embeds: payload.embeds ?? [],
attachments: payload.attachments,
message_reference: payload.message_reference
? {
...payload.message_reference,
message_id: BigInt(payload.message_reference.message_id),
channel_id: payload.message_reference.channel_id ? BigInt(payload.message_reference.channel_id) : undefined,
guild_id: payload.message_reference.guild_id ? BigInt(payload.message_reference.guild_id) : undefined,
}
: undefined,
allowed_mentions: payload.allowed_mentions
? {
parse: payload.allowed_mentions.parse,
users: payload.allowed_mentions.users?.map((value) => BigInt(value)),
roles: payload.allowed_mentions.roles?.map((value) => BigInt(value)),
replied_user: payload.allowed_mentions.replied_user,
}
: undefined,
flags: payload.flags,
nonce: payload.nonce,
favorite_meme_id: payload.favorite_meme_id ? BigInt(payload.favorite_meme_id) : undefined,
sticker_ids: payload.sticker_ids?.map((value) => BigInt(value)),
tts: payload.tts,
};
return message;
}
fluxer_api/src/api/test/TestConstants.ts
-17
View File
@@ -47,27 +47,10 @@ export const TEST_IDS = {
NONEXISTENT_MESSAGE: '999999999999999999',
NONEXISTENT_WEBHOOK: '999999999999999999',
} as const;
export const TEST_LIMITS = {
SCHEDULED_MESSAGE_MAX_DAYS: 30,
SCHEDULED_MESSAGE_MIN_DELAY_MS: 5 * 60 * 1000,
SCHEDULED_MESSAGE_MAX_DELAY_MS: 31 * 24 * 60 * 60 * 1000,
MFA_TICKET_SHORT_TTL: 1,
MFA_TICKET_LONG_TTL: 300,
PASSWORD_RESET_TOKEN_LENGTH: 64,
} as const;
export function generateUniquePassword(): string {
return `SecurePass-${Date.now()}!`;
}
export function generateFutureTimestamp(minutesInFuture = 5): string {
return new Date(Date.now() + minutesInFuture * 60 * 1000).toISOString();
}
export function generatePastTimestamp(hoursInPast = 1): string {
return new Date(Date.now() - hoursInPast * 60 * 60 * 1000).toISOString();
}
export function wait(ms: number): Promise<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
fluxer_api/src/api/test/TestHarnessController.ts
-44
View File
@@ -81,14 +81,12 @@ import {IpAuthorizationTokens, OAuth2AccessTokensByUser} from '../Tables';
import type {HonoApp, HonoEnv} from '../types/HonoEnv';
import {UserSearchRepository} from '../user/repositories/account/crud/UserSearchRepository';
import {AuthSessionRepository} from '../user/repositories/auth/AuthSessionRepository';
import {ScheduledMessageRepository} from '../user/repositories/ScheduledMessageRepository';
import {UserChannelRepository} from '../user/repositories/UserChannelRepository';
import {UserRepository} from '../user/repositories/UserRepository';
import {processUserDeletion} from '../user/services/UserDeletionService';
import {UserHarvestRepository} from '../user/UserHarvestRepository';
import {getExpiryBucket} from '../utils/AttachmentDecay';
import {parseReportedClientOs} from '../utils/SessionClientIdentity';
import {ScheduledMessageExecutor} from '../worker/executors/ScheduledMessageExecutor';
import {processExpiredAttachments} from '../worker/tasks/ExpireAttachments';
import {processInactivityDeletionsCore} from '../worker/tasks/ProcessInactivityDeletions';
import {setWorkerDependencies} from '../worker/WorkerContext';
@@ -1754,48 +1752,6 @@ export function TestHarnessController(app: HonoApp) {
await snowflakeService.shutdown();
}
});
app.post('/test/worker/send-scheduled-message/:userId/:scheduledMessageId', async (ctx) => {
ensureHarnessAccess(ctx);
const params = ctx.req.param() as {
userId?: string;
scheduledMessageId?: string;
};
const userIdParam = params.userId;
const scheduledMessageIdParam = params.scheduledMessageId;
if (!userIdParam) {
throw new Error('Missing userId parameter');
}
if (!scheduledMessageIdParam) {
throw new Error('Missing scheduledMessageId parameter');
}
const userId = createUserID(BigInt(userIdParam));
const scheduledMessageId = createMessageID(BigInt(scheduledMessageIdParam));
const snowflakeService = getSnowflakeService();
try {
const workerDeps = await initializeWorkerDepsWithHarnessEmail(ctx, snowflakeService);
setWorkerDependencies(workerDeps);
const scheduledMessageRepository = new ScheduledMessageRepository();
const scheduledMessage = await scheduledMessageRepository.getScheduledMessage(userId, scheduledMessageId);
if (!scheduledMessage) {
return ctx.json({success: false, reason: 'scheduled message not found'}, 404);
}
const logger = {
debug: (message: string, extra?: object) => Logger.debug(extra ?? {}, message),
info: (message: string, extra?: object) => Logger.info(extra ?? {}, message),
warn: (message: string, extra?: object) => Logger.warn(extra ?? {}, message),
error: (message: string, extra?: object) => Logger.error(extra ?? {}, message),
};
const executor = new ScheduledMessageExecutor(workerDeps, logger, scheduledMessageRepository);
await executor.execute({
userId: userId.toString(),
scheduledMessageId: scheduledMessageId.toString(),
expectedScheduledAt: scheduledMessage.scheduledAt.toISOString(),
});
return ctx.json({success: true}, 200);
} finally {
await snowflakeService.shutdown();
}
});
app.post('/test/attachment-decay/rows', async (ctx) => {
ensureHarnessAccess(ctx);
const payload = (await ctx.req.json()) as {
fluxer_api/src/api/types/HonoEnv.ts
-2
View File
@@ -18,7 +18,6 @@ import type {IChannelRepository} from '../channel/IChannelRepository';
import type {ChannelRequestService} from '../channel/services/ChannelRequestService';
import type {ChannelService} from '../channel/services/ChannelService';
import type {MessageRequestService} from '../channel/services/message/MessageRequestService';
import type {ScheduledMessageService} from '../channel/services/ScheduledMessageService';
import type {StreamPreviewService} from '../channel/services/StreamPreviewService';
import type {StreamService} from '../channel/services/StreamService';
import type {ConnectionRequestService} from '../connection/ConnectionRequestService';
@@ -177,7 +176,6 @@ export interface HonoEnv {
webhookService: WebhookService;
webhookRequestService: WebhookRequestService;
workerService: IWorkerService<WorkerTaskName>;
scheduledMessageService: ScheduledMessageService;
stripeService: StripeService;
ageVerificationService: AgeVerificationService;
applicationService: ApplicationService;
fluxer_api/src/api/user/controllers/UserController.ts
-2
View File
@@ -8,7 +8,6 @@ import {UserAuthController} from './UserAuthController';
import {UserChannelController} from './UserChannelController';
import {UserContentController} from './UserContentController';
import {UserRelationshipController} from './UserRelationshipController';
import {UserScheduledMessageController} from './UserScheduledMessageController';
export function UserController(app: HonoApp) {
UserAccountController(app);
@@ -16,7 +15,6 @@ export function UserController(app: HonoApp) {
UserRelationshipController(app);
UserChannelController(app);
UserContentController(app);
UserScheduledMessageController(app);
EntranceSoundController(app);
EntranceSoundPlayController(app);
}
fluxer_api/src/api/user/controllers/UserScheduledMessageController.ts
-141
View File
@@ -1,141 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import {UnknownMessageError} from '@fluxer/errors/src/domains/channel/UnknownMessageError';
import {ScheduledMessageIdParam} from '@fluxer/schema/src/domains/common/CommonParamSchemas';
import {
ScheduledMessageRequestSchema,
ScheduledMessageResponseSchema,
} from '@fluxer/schema/src/domains/message/ScheduledMessageSchemas';
import type {Context} from 'hono';
import {z} from 'zod';
import {createMessageID} from '../../BrandedTypes';
import {parseScheduledMessageInput} from '../../channel/controllers/ScheduledMessageParsing';
import {DefaultUserOnly, LoginRequired} from '../../middleware/AuthMiddleware';
import {RateLimitMiddleware} from '../../middleware/RateLimitMiddleware';
import {OpenAPI} from '../../middleware/ResponseTypeMiddleware';
import {RateLimitConfigs} from '../../RateLimitConfig';
import type {HonoApp, HonoEnv} from '../../types/HonoEnv';
import {Validator} from '../../Validator';
export function UserScheduledMessageController(app: HonoApp) {
app.get(
'/users/@me/scheduled-messages',
RateLimitMiddleware(RateLimitConfigs.USER_SAVED_MESSAGES_READ),
LoginRequired,
DefaultUserOnly,
OpenAPI({
operationId: 'list_scheduled_messages',
summary: 'List scheduled messages',
responseSchema: z.array(ScheduledMessageResponseSchema),
statusCode: 200,
security: ['bearerToken', 'sessionToken'],
tags: ['Users'],
description:
'Retrieves all scheduled messages for the current user. Returns list of messages that are scheduled to be sent at a future date and time.',
}),
async (ctx: Context<HonoEnv>) => {
const userId = ctx.get('user').id;
const scheduledMessageService = ctx.get('scheduledMessageService');
const scheduledMessages = await scheduledMessageService.listScheduledMessages(userId);
return ctx.json(
scheduledMessages.map((message) => message.toResponse()),
200,
);
},
);
app.get(
'/users/@me/scheduled-messages/:scheduled_message_id',
RateLimitMiddleware(RateLimitConfigs.USER_SAVED_MESSAGES_READ),
LoginRequired,
DefaultUserOnly,
Validator('param', ScheduledMessageIdParam),
OpenAPI({
operationId: 'get_scheduled_message',
summary: 'Get scheduled message',
responseSchema: ScheduledMessageResponseSchema,
statusCode: 200,
security: ['bearerToken', 'sessionToken'],
tags: ['Users'],
description:
'Retrieves details of a specific scheduled message by ID. Returns the message content, scheduled send time, and status.',
}),
async (ctx) => {
const userId = ctx.get('user').id;
const scheduledMessageId = createMessageID(ctx.req.valid('param').scheduled_message_id);
const scheduledMessageService = ctx.get('scheduledMessageService');
const scheduledMessage = await scheduledMessageService.getScheduledMessage(userId, scheduledMessageId);
if (!scheduledMessage) {
throw new UnknownMessageError();
}
return ctx.json(scheduledMessage.toResponse(), 200);
},
);
app.delete(
'/users/@me/scheduled-messages/:scheduled_message_id',
RateLimitMiddleware(RateLimitConfigs.USER_SAVED_MESSAGES_WRITE),
LoginRequired,
DefaultUserOnly,
Validator('param', ScheduledMessageIdParam),
OpenAPI({
operationId: 'cancel_scheduled_message',
summary: 'Cancel scheduled message',
responseSchema: null,
statusCode: 204,
security: ['bearerToken', 'sessionToken'],
tags: ['Users'],
description:
'Cancels and deletes a scheduled message before it is sent. The message will not be delivered if cancelled.',
}),
async (ctx) => {
const userId = ctx.get('user').id;
const scheduledMessageId = createMessageID(ctx.req.valid('param').scheduled_message_id);
const scheduledMessageService = ctx.get('scheduledMessageService');
await scheduledMessageService.cancelScheduledMessage(userId, scheduledMessageId);
return ctx.body(null, 204);
},
);
app.patch(
'/users/@me/scheduled-messages/:scheduled_message_id',
RateLimitMiddleware(RateLimitConfigs.USER_SAVED_MESSAGES_WRITE),
LoginRequired,
DefaultUserOnly,
Validator('param', ScheduledMessageIdParam),
OpenAPI({
operationId: 'update_scheduled_message',
summary: 'Update scheduled message',
responseSchema: ScheduledMessageResponseSchema,
requestSchema: ScheduledMessageRequestSchema,
requestFormSchema: ScheduledMessageRequestSchema,
statusCode: 200,
security: ['bearerToken', 'sessionToken'],
tags: ['Users'],
description:
'Updates an existing scheduled message before it is sent. Can modify message content, scheduled time, and timezone. Returns updated scheduled message details.',
}),
async (ctx) => {
const user = ctx.get('user');
const scheduledMessageService = ctx.get('scheduledMessageService');
const scheduledMessageId = createMessageID(ctx.req.valid('param').scheduled_message_id);
const existingMessage = await scheduledMessageService.getScheduledMessage(user.id, scheduledMessageId);
if (!existingMessage) {
throw new UnknownMessageError();
}
const channelId = existingMessage.channelId;
const {message, scheduledLocalAt, timezone} = await parseScheduledMessageInput({
ctx,
user,
channelId,
});
const scheduledMessage = await scheduledMessageService.updateScheduledMessage({
user,
channelId,
data: message,
scheduledLocalAt,
timezone,
scheduledMessageId,
existing: existingMessage,
});
return ctx.json(scheduledMessage.toResponse(), 200);
},
);
}
fluxer_api/src/api/user/repositories/ScheduledMessageRepository.ts
-65
View File
@@ -1,65 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import type {MessageID, UserID} from '../../BrandedTypes';
import {deleteOneOrMany, fetchMany, fetchOne, upsertOne} from '../../database/CassandraQueryExecution';
import {Db} from '../../database/CassandraTypes';
import type {ScheduledMessageRow} from '../../database/types/UserTypes';
import {ScheduledMessage} from '../../models/ScheduledMessage';
import {ScheduledMessages} from '../../Tables';
export class ScheduledMessageRepository {
private readonly fetchCql = ScheduledMessages.selectCql({
where: [ScheduledMessages.where.eq('user_id')],
});
async listScheduledMessages(userId: UserID, limit: number = 25): Promise<Array<ScheduledMessage>> {
const rows = await fetchMany<ScheduledMessageRow>(this.fetchCql, {
user_id: userId,
});
const messages = rows.map((row) => ScheduledMessage.fromRow(row));
return messages.sort((a, b) => (b.id > a.id ? 1 : a.id > b.id ? -1 : 0)).slice(0, limit);
}
async getScheduledMessage(userId: UserID, scheduledMessageId: MessageID): Promise<ScheduledMessage | null> {
const row = await fetchOne<ScheduledMessageRow>(
ScheduledMessages.selectCql({
where: [ScheduledMessages.where.eq('user_id'), ScheduledMessages.where.eq('scheduled_message_id')],
}),
{
user_id: userId,
scheduled_message_id: scheduledMessageId,
},
);
return row ? ScheduledMessage.fromRow(row) : null;
}
async upsertScheduledMessage(message: ScheduledMessage, ttlSeconds: number): Promise<void> {
await upsertOne(ScheduledMessages.upsertAllWithTtl(message.toRow(), ttlSeconds));
}
async deleteScheduledMessage(userId: UserID, scheduledMessageId: MessageID): Promise<void> {
await deleteOneOrMany(
ScheduledMessages.deleteByPk({
user_id: userId,
scheduled_message_id: scheduledMessageId,
}),
);
}
async markInvalid(userId: UserID, scheduledMessageId: MessageID, reason: string, ttlSeconds: number): Promise<void> {
await upsertOne(
ScheduledMessages.patchByPkWithTtl(
{
user_id: userId,
scheduled_message_id: scheduledMessageId,
},
{
status: Db.set('invalid'),
status_reason: Db.set(reason),
invalidated_at: Db.set(new Date()),
},
ttlSeconds,
),
);
}
}
fluxer_api/src/api/webhook/tests/WebhookPermissions.test.ts
+2 -3
View File
@@ -2,8 +2,7 @@
import {beforeAll, beforeEach, describe, it} from 'vitest';
import {createTestAccount} from '../../auth/tests/AuthTestUtils';
import {joinGuild} from '../../channel/tests/ScheduledMessageTestUtils';
import {createGuild} from '../../guild/tests/GuildTestUtils';
import {acceptInvite, createGuild} from '../../guild/tests/GuildTestUtils';
import {type ApiTestHarness, createApiTestHarness} from '../../test/ApiTestHarness';
import {HTTP_STATUS} from '../../test/TestConstants';
import {createBuilder} from '../../test/TestRequestBuilder';
@@ -23,7 +22,7 @@ describe('Webhook permissions', () => {
const guild = await createGuild(harness, owner.token, `Webhook Security ${Date.now()}`);
const channelId = guild.system_channel_id!;
const invite = await createChannelInvite(harness, owner.token, channelId);
await joinGuild(harness, member.token, invite.code);
await acceptInvite(harness, member.token, invite.code);
const webhook = await createWebhook(harness, channelId, owner.token, 'Test Webhook');
await createBuilder(harness, member.token)
.post(`/channels/${channelId}/webhooks`)
fluxer_api/src/api/worker/WorkerDependencies.ts
-5
View File
@@ -40,7 +40,6 @@ import type {IVoiceRoomStore} from '../infrastructure/IVoiceRoomStore';
import type {KVAccountDeletionQueueService} from '../infrastructure/KVAccountDeletionQueueService';
import type {KVActivityTracker} from '../infrastructure/KVActivityTracker';
import type {KVBulkMessageDeletionQueueService} from '../infrastructure/KVBulkMessageDeletionQueueService';
import type {KVScheduledJobQueueService} from '../infrastructure/KVScheduledJobQueueService';
import type {PremiumStateReconciliationQueueService} from '../infrastructure/PremiumStateReconciliationQueueService';
import type {UserCacheService} from '../infrastructure/UserCacheService';
import type {InstanceConfigRepository} from '../instance/InstanceConfigRepository';
@@ -87,7 +86,6 @@ import {
getKVAccountDeletionQueue,
getKVActivityTracker,
getKVBulkMessageDeletionQueue,
getKVScheduledJobQueue,
getLimitConfigService,
getNcmecSubmissionService,
getOAuth2TokenRepository,
@@ -164,7 +162,6 @@ export interface WorkerDependencies {
activityTracker: KVActivityTracker;
deletionQueueService: KVAccountDeletionQueueService;
bulkMessageDeletionQueueService: KVBulkMessageDeletionQueueService;
scheduledJobQueueService: KVScheduledJobQueueService;
premiumStateReconciliationQueueService: PremiumStateReconciliationQueueService;
deletionEligibilityService: UserDeletionEligibilityService;
voiceRoomStore: IVoiceRoomStore;
@@ -229,7 +226,6 @@ export async function initializeWorkerDependencies(snowflakeService: ISnowflakeS
const activityTracker = getKVActivityTracker();
const deletionQueueService = getKVAccountDeletionQueue();
const bulkMessageDeletionQueueService = getKVBulkMessageDeletionQueue();
const scheduledJobQueueService = getKVScheduledJobQueue();
const premiumStateReconciliationQueueService = getPremiumStateReconciliationQueueService();
const deletionEligibilityService = new UserDeletionEligibilityService(kvClient);
await ensureVoiceResourcesInitialized();
@@ -341,7 +337,6 @@ export async function initializeWorkerDependencies(snowflakeService: ISnowflakeS
activityTracker,
deletionQueueService,
bulkMessageDeletionQueueService,
scheduledJobQueueService,
premiumStateReconciliationQueueService,
deletionEligibilityService,
voiceRoomStore,
fluxer_api/src/api/worker/WorkerLaneConfig.ts
+1 -16
View File
@@ -32,7 +32,6 @@ const LANE_CONFIG = {
consumerName: 'workers_lifecycle',
tasks: [
'processStripeWebhook',
'sendScheduledMessage',
'sendSystemDm',
'userProcessPendingDeletion',
'userProcessPendingDeletions',
@@ -72,7 +71,6 @@ const LANE_CONFIG = {
'processInactivityDeletions',
'processPendingBulkMessageDeletions',
'processPremiumStateReconciliationQueue',
'processScheduledJobQueue',
'prunePostgresKvTtl',
'refreshSearchIndex',
'syncDiscoveryIndex',
@@ -214,13 +212,6 @@ function resolveWorkerLanes(config: WorkerLaneRuntimeConfig): Array<WorkerLaneDe
});
}
const SCHEDULED_JOB_DRAIN_INTERVALS_PER_ACK_WAIT = 2;
function resolveScheduledJobDrainIntervalSeconds(): number {
const minimumAckWaitMs = Math.min(...WORKER_LANES.map((lane) => lane.ackWaitMs));
return Math.max(1, Math.floor(minimumAckWaitMs / SCHEDULED_JOB_DRAIN_INTERVALS_PER_ACK_WAIT / 1000));
}
function resolveCronSchedulerEnabled(mode: APIWorkerMode, configuredValue: boolean | undefined): boolean {
if (configuredValue !== undefined) {
return configuredValue;
@@ -253,10 +244,4 @@ export function findLaneForTask(taskName: string): APIWorkerLaneName | null {
}
export type {WorkerLaneDefinition};
export {
resolveCronSchedulerEnabled,
resolveScheduledJobDrainIntervalSeconds,
resolveWorkerLanes,
validateLaneCompleteness,
WORKER_LANES,
};
export {resolveCronSchedulerEnabled, resolveWorkerLanes, validateLaneCompleteness, WORKER_LANES};
fluxer_api/src/api/worker/WorkerMain.ts
-9
View File
@@ -25,7 +25,6 @@ import {clearWorkerDependencies, setWorkerDependencies} from './WorkerContext';
import {initializeWorkerDependencies, shutdownWorkerDependencies, type WorkerDependencies} from './WorkerDependencies';
import {
resolveCronSchedulerEnabled,
resolveScheduledJobDrainIntervalSeconds,
resolveWorkerLanes,
validateLaneCompleteness,
type WorkerLaneDefinition,
@@ -54,13 +53,6 @@ function registerCronJobs(cron: CronScheduler): void {
cron.upsert('processPremiumStateReconciliationQueue', 'processPremiumStateReconciliationQueue', {}, '0 * * * * *', {
ledger: false,
});
cron.upsert(
'processScheduledJobQueue',
'processScheduledJobQueue',
{},
`*/${resolveScheduledJobDrainIntervalSeconds()} * * * * *`,
{ledger: false},
);
if (!Config.instance.selfHosted) {
cron.upsert('processExpiredPremiumSweep', 'processExpiredPremiumSweep', {}, '0 0 * * * *', {ledger: false});
}
@@ -249,7 +241,6 @@ export async function startWorkerMain(): Promise<void> {
const runner = new WorkerRunner({
tasks: laneTasks,
queue,
scheduledJobQueue: dependencies.scheduledJobQueueService,
consumerName: lane.consumerName,
laneName: lane.name,
ledger: jobLedger,
fluxer_api/src/api/worker/WorkerRunner.ts
+5 -39
View File
@@ -4,7 +4,6 @@ import {randomUUID} from 'node:crypto';
import type {IWorkerService} from '@pkgs/worker/src/contracts/IWorkerService';
import {JobCancelledError, type WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
import type {ConsumerMessages, JsMsg} from 'nats';
import {buildScheduledJobIdentity, type ParkedScheduledJob} from '../infrastructure/KVScheduledJobQueueService';
import type {IJobLedgerRepository} from '../jobs/IJobLedgerRepository';
import {Logger} from '../Logger';
import {getWorkerService} from '../middleware/ServiceRegistry';
@@ -42,14 +41,9 @@ interface WorkerRunnerQueue {
publishToDlq(taskType: string, originalPayload: Record<string, unknown>, meta: WorkerRunnerDlqMeta): Promise<void>;
}
interface WorkerRunnerScheduledJobQueue {
parkJob(job: ParkedScheduledJob, releaseAt: Date): Promise<void>;
}
interface WorkerRunnerOptions {
tasks: Record<string, WorkerTaskHandler>;
queue: WorkerRunnerQueue;
scheduledJobQueue: WorkerRunnerScheduledJobQueue;
consumerName: string;
laneName: string;
ledger: IJobLedgerRepository;
@@ -62,7 +56,6 @@ interface WorkerRunnerOptions {
export class WorkerRunner {
private readonly tasks: Record<string, WorkerTaskHandler>;
private readonly queue: WorkerRunnerQueue;
private readonly scheduledJobQueue: WorkerRunnerScheduledJobQueue;
private readonly consumerName: string;
private readonly laneName: string;
private readonly workerId: string;
@@ -78,7 +71,6 @@ export class WorkerRunner {
constructor(options: WorkerRunnerOptions) {
this.tasks = options.tasks;
this.queue = options.queue;
this.scheduledJobQueue = options.scheduledJobQueue;
this.consumerName = options.consumerName;
this.laneName = options.laneName;
this.workerId = options.workerId ?? `worker-${options.laneName}-${randomUUID()}`;
@@ -201,37 +193,11 @@ export class WorkerRunner {
if (Number.isFinite(runAtMs)) {
const delayMs = runAtMs - Date.now();
if (delayMs > 0) {
const deliveryCount = msg.info.deliveryCount;
const shouldPark = delayMs > this.ackWaitMs || deliveryCount >= this.maxDeliver - 1;
if (shouldPark) {
const ledgerJobIdValue = ledgerJobId === null ? null : ledgerJobId.toString();
try {
await this.scheduledJobQueue.parkJob(
{
jobIdentity: buildScheduledJobIdentity(taskType, jobPayload, runAtMs, ledgerJobIdValue),
taskType,
payload: jobPayload,
runAtMs,
ledgerJobId: ledgerJobIdValue,
},
new Date(runAtMs - this.ackWaitMs),
);
msg.ack();
Logger.debug(
{taskType, seq: msg.seq, runAt, deliveryCount},
'Parked scheduled job in the due queue to free ack slot',
);
} catch (error) {
Logger.error({taskType, seq: msg.seq, err: error}, 'Failed to park scheduled job, falling back to NAK');
msg.nak(Math.min(delayMs, this.ackWaitMs - 5000));
}
} else {
Logger.debug(
{taskType, seq: msg.seq, runAt, delayMs},
'Job scheduled for future execution, redelivering with delay',
);
msg.nak(delayMs);
}
Logger.debug(
{taskType, seq: msg.seq, runAt, delayMs},
'Job scheduled for future execution, redelivering with delay',
);
msg.nak(delayMs);
return false;
}
}
fluxer_api/src/api/worker/WorkerTaskRegistry.ts
-4
View File
@@ -30,13 +30,11 @@ import processExpiredPremiumSweep from './tasks/ProcessExpiredPremiumSweep';
import processInactivityDeletions from './tasks/ProcessInactivityDeletions';
import processPendingBulkMessageDeletions from './tasks/ProcessPendingBulkMessageDeletions';
import processPremiumStateReconciliationQueue from './tasks/ProcessPremiumStateReconciliationQueue';
import processScheduledJobQueue from './tasks/ProcessScheduledJobQueue';
import processStripeWebhook from './tasks/ProcessStripeWebhook';
import prunePostgresKvTtl from './tasks/PrunePostgresKvTtl';
import reconcileUserPayments from './tasks/ReconcileUserPayments';
import refreshSearchIndex from './tasks/RefreshSearchIndex';
import revalidateUserConnections from './tasks/RevalidateUserConnections';
import {sendScheduledMessage} from './tasks/SendScheduledMessage';
import {sendSystemDm} from './tasks/SendSystemDm';
import syncDiscoveryIndex from './tasks/SyncDiscoveryIndex';
import syncDisposableEmailDomains from './tasks/SyncDisposableEmailDomains';
@@ -76,12 +74,10 @@ export const workerTasks: Record<WorkerTaskName, WorkerTaskHandler> = {
processInactivityDeletions,
processPendingBulkMessageDeletions,
processPremiumStateReconciliationQueue,
processScheduledJobQueue,
reconcileUserPayments,
prunePostgresKvTtl,
refreshSearchIndex,
revalidateUserConnections,
sendScheduledMessage,
sendSystemDm,
syncFileShaBlocklists,
syncUrlBlocklists,
fluxer_api/src/api/worker/executors/ScheduledMessageExecutor.ts
-125
View File
@@ -1,125 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import type {MessageID, UserID} from '../../BrandedTypes';
import {createMessageID, createUserID} from '../../BrandedTypes';
import {SCHEDULED_MESSAGE_TTL_SECONDS} from '../../channel/services/ScheduledMessageService';
import {createRequestCache} from '../../middleware/RequestCacheMiddleware';
import {ScheduledMessageRepository} from '../../user/repositories/ScheduledMessageRepository';
import type {WorkerDependencies} from '../WorkerDependencies';
interface WorkerLogger {
debug(message: string, extra?: object): void;
info(message: string, extra?: object): void;
warn(message: string, extra?: object): void;
error(message: string, extra?: object): void;
}
export interface SendScheduledMessageParams {
userId: string;
scheduledMessageId: string;
expectedScheduledAt: string;
}
export class ScheduledMessageExecutor {
constructor(
private readonly deps: WorkerDependencies,
private readonly logger: WorkerLogger,
private readonly scheduledMessageRepository: ScheduledMessageRepository = new ScheduledMessageRepository(),
) {}
async execute(params: SendScheduledMessageParams): Promise<void> {
const userId = this.parseUserID(params.userId);
const scheduledMessageId = this.parseMessageID(params.scheduledMessageId);
if (!userId || !scheduledMessageId) {
this.logger.warn('Malformed scheduled message job payload', {payload: params});
return;
}
const expectedScheduledAt = this.parseScheduledAt(params.expectedScheduledAt);
if (!expectedScheduledAt) {
this.logger.warn('Invalid expectedScheduledAt for scheduled message job', {payload: params});
return;
}
const scheduledMessage = await this.scheduledMessageRepository.getScheduledMessage(userId, scheduledMessageId);
if (!scheduledMessage) {
this.logger.info('Scheduled message not found, skipping', {userId, scheduledMessageId});
return;
}
if (scheduledMessage.status !== 'pending') {
this.logger.info('Scheduled message already processed', {
scheduledMessageId,
status: scheduledMessage.status,
});
return;
}
if (scheduledMessage.scheduledAt.toISOString() !== expectedScheduledAt.toISOString()) {
this.logger.info('Scheduled message time mismatch, skipping stale job', {
scheduledMessageId,
expected: expectedScheduledAt.toISOString(),
actual: scheduledMessage.scheduledAt.toISOString(),
});
return;
}
const user = await this.deps.userRepository.findUnique(userId);
if (!user) {
await this.markInvalid(userId, scheduledMessageId, 'User not found');
return;
}
const messageRequest = scheduledMessage.parseToMessageRequest();
const requestCache = createRequestCache();
try {
await this.deps.channelService.messages.send.validateMessageCanBeSent({
user,
channelId: scheduledMessage.channelId,
data: messageRequest,
});
await this.deps.channelService.messages.send.sendMessage({
user,
channelId: scheduledMessage.channelId,
data: messageRequest,
requestCache,
});
await this.scheduledMessageRepository.deleteScheduledMessage(userId, scheduledMessageId);
this.logger.info('Scheduled message sent successfully', {scheduledMessageId, userId});
} catch (error) {
const reason = error instanceof Error ? error.message : 'Failed to send scheduled message';
this.logger.warn('Marking scheduled message invalid', {scheduledMessageId, userId, reason});
await this.markInvalid(userId, scheduledMessageId, reason);
} finally {
requestCache.clear();
}
}
private async markInvalid(userId: UserID, scheduledMessageId: MessageID, reason: string): Promise<void> {
try {
await this.scheduledMessageRepository.markInvalid(
userId,
scheduledMessageId,
reason,
SCHEDULED_MESSAGE_TTL_SECONDS,
);
} catch (error) {
this.logger.error('Failed to mark scheduled message invalid', {error, scheduledMessageId});
}
}
private parseUserID(value: string): UserID | null {
try {
return createUserID(BigInt(value));
} catch {
return null;
}
}
private parseMessageID(value: string): MessageID | null {
try {
return createMessageID(BigInt(value));
} catch {
return null;
}
}
private parseScheduledAt(value: string): Date | null {
const date = new Date(value);
return Number.isNaN(date.getTime()) ? null : date;
}
}
fluxer_api/src/api/worker/tasks/ProcessScheduledJobQueue.ts
-80
View File
@@ -1,80 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import type {WorkerTaskHandler} from '@pkgs/worker/src/contracts/WorkerTask';
import type {WorkerJobPayload} from '@pkgs/worker/src/contracts/WorkerTypes';
import type {ParkedScheduledJob} from '../../infrastructure/KVScheduledJobQueueService';
import {Logger} from '../../Logger';
import {getWorkerDependencies} from '../WorkerContext';
import {findLaneForTask, type WorkerTaskName} from '../WorkerLaneConfig';
const MAX_JOBS_PER_RUN = 500;
function buildReleasePayload(job: ParkedScheduledJob): WorkerJobPayload {
if (job.ledgerJobId === null) {
return job.payload;
}
return {...job.payload, __jobId: job.ledgerJobId};
}
const processScheduledJobQueue: WorkerTaskHandler = async (_payload, helpers) => {
const {scheduledJobQueueService, workerService} = getWorkerDependencies();
const nowMs = Date.now();
const readyJobs = await scheduledJobQueueService.getReadyJobs(nowMs, MAX_JOBS_PER_RUN);
if (readyJobs.length === 0) {
helpers.logger.debug('No parked scheduled jobs are due');
return;
}
let releasedCount = 0;
let claimedElsewhereCount = 0;
let droppedCount = 0;
let failedCount = 0;
for (const {job, member} of readyJobs) {
try {
const claimed = await scheduledJobQueueService.claimJob(job.jobIdentity, member);
if (!claimed) {
claimedElsewhereCount += 1;
continue;
}
if (findLaneForTask(job.taskType) === null) {
droppedCount += 1;
Logger.error(
{jobIdentity: job.jobIdentity, taskType: job.taskType},
'Dropping parked scheduled job with an unknown task type',
);
continue;
}
await workerService.addJob(job.taskType as WorkerTaskName, buildReleasePayload(job), {
runAt: new Date(job.runAtMs),
jobKey: job.jobIdentity,
skipLedger: true,
});
releasedCount += 1;
} catch (error) {
failedCount += 1;
Logger.error(
{error, jobIdentity: job.jobIdentity, taskType: job.taskType},
'Failed to release parked scheduled job, re-parking for the next drain',
);
try {
await scheduledJobQueueService.parkJob(job, new Date(nowMs));
} catch (parkError) {
Logger.error(
{error: parkError, jobIdentity: job.jobIdentity, taskType: job.taskType},
'Failed to re-park scheduled job after a release failure',
);
}
}
}
helpers.logger.info(
{
total: readyJobs.length,
releasedCount,
claimedElsewhereCount,
droppedCount,
failedCount,
},
'Released parked scheduled jobs',
);
};
export default processScheduledJobQueue;
fluxer_api/src/api/worker/tasks/SendScheduledMessage.ts
-20
View File
@@ -1,20 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import type {WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
import {z} from 'zod';
import {ScheduledMessageExecutor, type SendScheduledMessageParams} from '../executors/ScheduledMessageExecutor';
import {getWorkerDependencies} from '../WorkerContext';
const PayloadSchema = z.object({
userId: z.string(),
scheduledMessageId: z.string(),
expectedScheduledAt: z.string(),
});
export async function sendScheduledMessage(payload: unknown, helpers: WorkerTaskHelpers): Promise<void> {
const validated = PayloadSchema.parse(payload) as SendScheduledMessageParams;
const deps = getWorkerDependencies();
const executor = new ScheduledMessageExecutor(deps, helpers.logger);
const result = await executor.execute(validated);
return result;
}
fluxer_api/src/api/worker/tests/ScheduledJobQueue.test.ts
-343
View File
@@ -1,343 +0,0 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import type {IWorkerService} from '@pkgs/worker/src/contracts/IWorkerService';
import type {WorkerTaskHelpers} from '@pkgs/worker/src/contracts/WorkerTask';
import type {WorkerJobOptions, WorkerJobPayload} from '@pkgs/worker/src/contracts/WorkerTypes';
import type {JsMsg} from 'nats';
import {afterEach, beforeAll, describe, expect, it, vi} from 'vitest';
import {KVScheduledJobQueueService} from '../../infrastructure/KVScheduledJobQueueService';
import type {IJobLedgerRepository} from '../../jobs/IJobLedgerRepository';
import {setInjectedWorkerService} from '../../middleware/ServiceRegistry';
import {MockKVProvider} from '../../test/mocks/MockKVProvider';
import {NoopLogger} from '../../test/mocks/NoopLogger';
import {NoopWorkerService} from '../../test/NoopWorkerService';
import processScheduledJobQueue from '../tasks/ProcessScheduledJobQueue';
import {setWorkerDependenciesForTest} from '../WorkerContext';
import {WorkerRunner} from '../WorkerRunner';
const LANE_ACK_WAIT_MS = 60000;
const LANE_MAX_DELIVER = 25;
const TASK_TYPE = 'sendScheduledMessage';
interface RecordedJob {
taskType: string;
payload: WorkerJobPayload;
options: WorkerJobOptions | undefined;
}
class RecordingWorkerService implements IWorkerService {
readonly jobs: Array<RecordedJob> = [];
async addJob<TPayload extends WorkerJobPayload = WorkerJobPayload>(
taskType: string,
payload: TPayload,
options?: WorkerJobOptions,
): Promise<bigint> {
this.jobs.push({taskType, payload, options});
return BigInt(this.jobs.length);
}
async cancelJob(_jobId: bigint): Promise<boolean> {
return false;
}
async retryDeadLetterJob(_jobId: bigint): Promise<boolean> {
return false;
}
}
const queueStub = {
getConnectionManager: () => {
throw new Error('WorkerRunner tests never consume messages');
},
getStreamName: () => 'JOBS',
enqueue: vi.fn(),
publishToDlq: vi.fn(),
};
class ConcurrentlyClaimedKVProvider extends MockKVProvider {
override async removeBulkDeletion(queueKey: string, secondaryKey: string): Promise<boolean> {
await super.removeBulkDeletion(queueKey, secondaryKey);
return false;
}
}
class TestWorkerRunner extends WorkerRunner {
async runJob(taskType: string, msg: JsMsg): Promise<boolean> {
return await this.processJob(taskType, msg);
}
}
function createRunner(scheduledJobQueue: KVScheduledJobQueueService): TestWorkerRunner {
return new TestWorkerRunner({
tasks: {[TASK_TYPE]: async () => {}},
queue: queueStub,
scheduledJobQueue,
consumerName: 'workers_lifecycle',
laneName: 'lifecycle',
ledger: {} as IJobLedgerRepository,
concurrency: 8,
maxDeliver: LANE_MAX_DELIVER,
ackWaitMs: LANE_ACK_WAIT_MS,
});
}
function createJobMessage(options: {
runAt: Date;
payload: Record<string, unknown>;
seq?: number;
deliveryCount?: number;
}) {
const envelope = {
payload: options.payload,
run_at: options.runAt.toISOString(),
max_attempts: 5,
priority: 0,
created_at: new Date().toISOString(),
};
return {
seq: options.seq ?? 1,
subject: `jobs.${TASK_TYPE}`,
redelivered: (options.deliveryCount ?? 1) > 1,
data: new TextEncoder().encode(JSON.stringify(envelope)),
info: {deliveryCount: options.deliveryCount ?? 1},
ack: vi.fn(),
nak: vi.fn(),
term: vi.fn(),
};
}
function createHelpers(): WorkerTaskHelpers {
return {
logger: new NoopLogger(),
jobId: 0n,
addJob: async () => 0n,
reportProgress: async () => {},
shouldCancel: async () => false,
setContextLink: async () => {},
};
}
async function drain(scheduledJobQueue: KVScheduledJobQueueService, workerService: RecordingWorkerService) {
setWorkerDependenciesForTest({scheduledJobQueueService: scheduledJobQueue, workerService});
await processScheduledJobQueue({}, createHelpers());
}
describe('Scheduled job due queue', () => {
beforeAll(() => {
setInjectedWorkerService(new NoopWorkerService());
});
afterEach(() => {
vi.useRealTimers();
queueStub.enqueue.mockClear();
queueStub.publishToDlq.mockClear();
});
it('naks a job due inside one ack-wait instead of parking it', async () => {
const scheduledJobQueue = new KVScheduledJobQueueService(new MockKVProvider());
const runner = createRunner(scheduledJobQueue);
const msg = createJobMessage({
runAt: new Date(Date.now() + LANE_ACK_WAIT_MS - 30000),
payload: {scheduledMessageId: '77', __jobId: '4242'},
});
await runner.runJob(TASK_TYPE, msg as unknown as JsMsg);
expect(msg.nak).toHaveBeenCalledTimes(1);
expect(msg.ack).not.toHaveBeenCalled();
const delayMs = msg.nak.mock.calls[0]![0] as number;
expect(delayMs).toBeGreaterThan(0);
expect(delayMs).toBeLessThanOrEqual(LANE_ACK_WAIT_MS - 30000);
expect(await scheduledJobQueue.getQueueSize()).toBe(0);
});
it('parks a job due beyond one ack-wait without republishing it', async () => {
const scheduledJobQueue = new KVScheduledJobQueueService(new MockKVProvider());
const runner = createRunner(scheduledJobQueue);
const runAt = new Date(Date.now() + 60 * 60 * 1000);
const msg = createJobMessage({runAt, payload: {scheduledMessageId: '77', __jobId: '4242'}});
await runner.runJob(TASK_TYPE, msg as unknown as JsMsg);
expect(msg.ack).toHaveBeenCalledTimes(1);
expect(msg.nak).not.toHaveBeenCalled();
expect(queueStub.enqueue).not.toHaveBeenCalled();
expect(await scheduledJobQueue.getQueueSize()).toBe(1);
expect(await scheduledJobQueue.getReadyJobs(Date.now(), 10)).toHaveLength(0);
const due = await scheduledJobQueue.getReadyJobs(runAt.getTime() - LANE_ACK_WAIT_MS, 10);
expect(due).toHaveLength(1);
expect(due[0]!.job).toEqual({
jobIdentity: '4242',
taskType: TASK_TYPE,
payload: {scheduledMessageId: '77'},
runAtMs: runAt.getTime(),
ledgerJobId: '4242',
});
expect(due[0]!.member).toContain('"jobIdentity":"4242"');
});
it('releases a parked job exactly once when its due time arrives', async () => {
vi.useFakeTimers({toFake: ['Date']});
const start = new Date('2026-08-30T12:00:00.000Z');
vi.setSystemTime(start);
const scheduledJobQueue = new KVScheduledJobQueueService(new MockKVProvider());
const runner = createRunner(scheduledJobQueue);
const runAt = new Date(start.getTime() + 60 * 60 * 1000);
const msg = createJobMessage({runAt, payload: {scheduledMessageId: '77', __jobId: '4242'}});
await runner.runJob(TASK_TYPE, msg as unknown as JsMsg);
const workerService = new RecordingWorkerService();
await drain(scheduledJobQueue, workerService);
expect(workerService.jobs).toHaveLength(0);
vi.setSystemTime(new Date(runAt.getTime() - LANE_ACK_WAIT_MS + 1000));
await drain(scheduledJobQueue, workerService);
expect(workerService.jobs).toHaveLength(1);
expect(workerService.jobs[0]!.taskType).toBe(TASK_TYPE);
expect(workerService.jobs[0]!.payload).toEqual({scheduledMessageId: '77', __jobId: '4242'});
expect(workerService.jobs[0]!.options?.runAt?.getTime()).toBe(runAt.getTime());
expect(workerService.jobs[0]!.options?.jobKey).toBe('4242');
expect(workerService.jobs[0]!.options?.skipLedger).toBe(true);
expect(await scheduledJobQueue.getQueueSize()).toBe(0);
await drain(scheduledJobQueue, workerService);
expect(workerService.jobs).toHaveLength(1);
});
it('runs a job parked twice only once', async () => {
vi.useFakeTimers({toFake: ['Date']});
const start = new Date('2026-08-30T12:00:00.000Z');
vi.setSystemTime(start);
const scheduledJobQueue = new KVScheduledJobQueueService(new MockKVProvider());
const runner = createRunner(scheduledJobQueue);
const runAt = new Date(start.getTime() + 60 * 60 * 1000);
const payload = {scheduledMessageId: '77', __jobId: '4242'};
await runner.runJob(TASK_TYPE, createJobMessage({runAt, payload, seq: 1}) as unknown as JsMsg);
await runner.runJob(TASK_TYPE, createJobMessage({runAt, payload, seq: 2, deliveryCount: 2}) as unknown as JsMsg);
expect(await scheduledJobQueue.getQueueSize()).toBe(1);
vi.setSystemTime(new Date(runAt.getTime() - LANE_ACK_WAIT_MS + 1000));
const workerService = new RecordingWorkerService();
await drain(scheduledJobQueue, workerService);
expect(workerService.jobs).toHaveLength(1);
expect(await scheduledJobQueue.getQueueSize()).toBe(0);
});
it('parks a nearly delivery-exhausted job so it gets a fresh delivery budget', async () => {
const scheduledJobQueue = new KVScheduledJobQueueService(new MockKVProvider());
const runner = createRunner(scheduledJobQueue);
const runAt = new Date(Date.now() + 20000);
const msg = createJobMessage({
runAt,
payload: {scheduledMessageId: '77', __jobId: '4242'},
deliveryCount: LANE_MAX_DELIVER - 1,
});
await runner.runJob(TASK_TYPE, msg as unknown as JsMsg);
expect(msg.ack).toHaveBeenCalledTimes(1);
expect(msg.nak).not.toHaveBeenCalled();
expect(queueStub.enqueue).not.toHaveBeenCalled();
const workerService = new RecordingWorkerService();
await drain(scheduledJobQueue, workerService);
expect(workerService.jobs).toHaveLength(1);
expect(workerService.jobs[0]!.options?.runAt?.getTime()).toBe(runAt.getTime());
});
it('does nothing when the due queue is empty', async () => {
const scheduledJobQueue = new KVScheduledJobQueueService(new MockKVProvider());
const workerService = new RecordingWorkerService();
await expect(drain(scheduledJobQueue, workerService)).resolves.toBeUndefined();
expect(workerService.jobs).toHaveLength(0);
expect(await scheduledJobQueue.getQueueSize()).toBe(0);
});
it('does not release a job another drain claimed first', async () => {
const scheduledJobQueue = new KVScheduledJobQueueService(new ConcurrentlyClaimedKVProvider());
const runner = createRunner(scheduledJobQueue);
const runAt = new Date(Date.now() + 20000);
const msg = createJobMessage({
runAt,
payload: {scheduledMessageId: '77', __jobId: '4242'},
deliveryCount: LANE_MAX_DELIVER - 1,
});
await runner.runJob(TASK_TYPE, msg as unknown as JsMsg);
expect(await scheduledJobQueue.getReadyJobs(Date.now(), 10)).toHaveLength(1);
const workerService = new RecordingWorkerService();
await drain(scheduledJobQueue, workerService);
expect(workerService.jobs).toHaveLength(0);
expect(await scheduledJobQueue.getQueueSize()).toBe(0);
});
});
describe('Parked job claim survives secondary key loss', () => {
it('claims by member when the secondary key was evicted', async () => {
const kv = new MockKVProvider();
const service = new KVScheduledJobQueueService(kv);
const runAtMs = Date.now() + 86_400_000;
const job = {
jobIdentity: 'job-evicted',
taskType: TASK_TYPE,
payload: {message_id: '1'} as WorkerJobPayload,
runAtMs,
ledgerJobId: null,
};
await service.parkJob(job, new Date(Date.now() - 1000));
expect(await service.getQueueSize()).toBe(1);
await kv.del('scheduled_job_queue:job-evicted');
const ready = await service.getReadyJobs(Date.now(), 10);
expect(ready).toHaveLength(1);
expect(await service.claimJob(ready[0]!.job.jobIdentity, ready[0]!.member)).toBe(true);
expect(await service.getQueueSize()).toBe(0);
});
it('leaves the entry wedged when claiming without the member', async () => {
const kv = new MockKVProvider();
const service = new KVScheduledJobQueueService(kv);
const job = {
jobIdentity: 'job-wedged',
taskType: TASK_TYPE,
payload: {message_id: '2'} as WorkerJobPayload,
runAtMs: Date.now() + 86_400_000,
ledgerJobId: null,
};
await service.parkJob(job, new Date(Date.now() - 1000));
await kv.del('scheduled_job_queue:job-wedged');
expect(await service.claimJob('job-wedged')).toBe(false);
expect(await service.getQueueSize()).toBe(1);
});
it('still reports a second claim of the same member as already claimed', async () => {
const kv = new MockKVProvider();
const service = new KVScheduledJobQueueService(kv);
const job = {
jobIdentity: 'job-twice',
taskType: TASK_TYPE,
payload: {message_id: '3'} as WorkerJobPayload,
runAtMs: Date.now() + 86_400_000,
ledgerJobId: null,
};
await service.parkJob(job, new Date(Date.now() - 1000));
const ready = await service.getReadyJobs(Date.now(), 10);
const member = ready[0]!.member;
expect(await service.claimJob('job-twice', member)).toBe(true);
expect(await service.claimJob('job-twice', member)).toBe(false);
});
});
fluxer_api/src/api/worker/tests/WorkerAckHeartbeat.test.ts
+3 -8
View File
@@ -3,10 +3,8 @@
import {JobCancelledError} from '@pkgs/worker/src/contracts/WorkerTask';
import type {JsMsg} from 'nats';
import {afterEach, beforeAll, describe, expect, it, vi} from 'vitest';
import {KVScheduledJobQueueService} from '../../infrastructure/KVScheduledJobQueueService';
import type {IJobLedgerRepository} from '../../jobs/IJobLedgerRepository';
import {setInjectedWorkerService} from '../../middleware/ServiceRegistry';
import {MockKVProvider} from '../../test/mocks/MockKVProvider';
import {NoopWorkerService} from '../../test/NoopWorkerService';
import {WorkerRunner} from '../WorkerRunner';
@@ -32,7 +30,6 @@ function createRunner(task: () => Promise<void>): TestWorkerRunner {
return new TestWorkerRunner({
tasks: {[TASK_TYPE]: task},
queue: queueStub,
scheduledJobQueue: new KVScheduledJobQueueService(new MockKVProvider()),
consumerName: 'workers_batch',
laneName: 'batch',
ledger: {} as IJobLedgerRepository,
@@ -146,12 +143,10 @@ describe('Worker ack heartbeat', () => {
expect(msg.working.mock.calls.length).toBe(heartbeatsAtCompletion);
});
it('does not heartbeat a parked scheduled job', async () => {
const scheduledJobQueue = new KVScheduledJobQueueService(new MockKVProvider());
it('does not heartbeat a job deferred to the future', async () => {
const runner = new TestWorkerRunner({
tasks: {[TASK_TYPE]: async () => {}},
queue: queueStub,
scheduledJobQueue,
consumerName: 'workers_batch',
laneName: 'batch',
ledger: {} as IJobLedgerRepository,
@@ -174,8 +169,8 @@ describe('Worker ack heartbeat', () => {
await runner.runJob(TASK_TYPE, msg as unknown as JsMsg);
expect(msg.ack).toHaveBeenCalledTimes(1);
expect(msg.ack).not.toHaveBeenCalled();
expect(msg.nak).toHaveBeenCalledTimes(1);
expect(msg.working).not.toHaveBeenCalled();
expect(await scheduledJobQueue.getQueueSize()).toBe(1);
});
});
fluxer_api/src/api/worker/tests/WorkerLaneConfig.test.ts
+1 -12
View File
@@ -1,12 +1,7 @@
// SPDX-License-Identifier: AGPL-3.0-or-later
import {describe, expect, it} from 'vitest';
import {
resolveCronSchedulerEnabled,
resolveScheduledJobDrainIntervalSeconds,
resolveWorkerLanes,
WORKER_LANES,
} from '../WorkerLaneConfig';
import {resolveCronSchedulerEnabled, resolveWorkerLanes, WORKER_LANES} from '../WorkerLaneConfig';
describe('WorkerLaneConfig', () => {
it('returns all lanes in all_lanes mode', () => {
@@ -102,10 +97,4 @@ describe('WorkerLaneConfig', () => {
expect(resolveCronSchedulerEnabled('single_lane', true)).toBe(true);
expect(resolveCronSchedulerEnabled('all_lanes', false)).toBe(false);
});
it('drains parked scheduled jobs well inside the shortest lane ack-wait', () => {
const intervalSeconds = resolveScheduledJobDrainIntervalSeconds();
const shortestAckWaitMs = Math.min(...WORKER_LANES.map((lane) => lane.ackWaitMs));
expect(intervalSeconds).toBeGreaterThanOrEqual(1);
expect(intervalSeconds * 1000 * 2).toBeLessThanOrEqual(shortestAckWaitMs);
});
});