wip: notifications

api/src/core/Dispatcher.ts

@@ -0,0 +1,4 @@
+export interface Dispatcher {
+    register(): void;
+    registerWorkers(): Promise<void>;
+}

api/src/core/EventBus.ts

@@ -3,10 +3,10 @@ import type { TasksSchemaTypeForSelect } from 'taskview-db-schemas';
 import { $logger } from '../modules/logget';
 
 export interface AppEvents {
-    'task.created': { task: TasksSchemaTypeForSelect; userId: number };
-    'task.updated': { task: TasksSchemaTypeForSelect; changes: Record<string, unknown>; userId: number };
-    'task.assigneesChanged': { taskId: number; userIds: number[] };
-    'task.deleted': { taskId: number; goalId: number };
+    'task.created': { task: TasksSchemaTypeForSelect; initiatorId: number };
+    'task.updated': { task: TasksSchemaTypeForSelect; changes: Record<string, unknown>; initiatorId: number };
+    'task.assigneesChanged': { taskId: number; userIds: number[]; initiatorId: number };
+    'task.deleted': { taskId: number; goalId: number; initiatorId: number };
 }
 
 type EventName = keyof AppEvents;

api/src/core/JobQueue.ts

@@ -1,5 +1,6 @@
 import { PgBoss } from 'pg-boss';
 import { $logger } from '../modules/logget';
+import { Database } from '../modules/db';
 
 let boss: PgBoss | null = null;
 
@@ -29,3 +30,24 @@ export function getJobQueue(): PgBoss {
     }
     return boss;
 }
+
+/** Cancel jobs by singletonKey — finds and deletes matching queued jobs */
+export async function cancelJobBySingletonKey(queueName: string, singletonKey: string): Promise<void> {
+    if (!boss) return;
+    const db = Database.getInstance();
+    const result = await db.query<{ id: string }>(
+        `SELECT id FROM pgboss.job WHERE name = $1 AND singleton_key = $2 AND state IN ('created', 'retry')`,
+        [queueName, singletonKey],
+    );
+
+    const count = result?.rows?.length ?? 0;
+    if (count === 0) {
+        $logger.info(`[JobQueue] Cancel: no jobs found for key="${singletonKey}"`);
+        return;
+    }
+
+    for (const row of result!.rows) {
+        await boss.deleteJob(queueName, row.id);
+        $logger.info(`[JobQueue] Deleted job id=${row.id} key="${singletonKey}"`);
+    }
+}

api/src/core/all-events.ts

@@ -1,17 +1,20 @@
 import { startJobQueue } from './JobQueue';
-import { registerNotificationEventHandlers, registerNotificationWorkers } from '../tv-modules/notifications/notifications.events';
+import type { Dispatcher } from './Dispatcher';
+import { NotificationDispatcher } from '../tv-modules/notifications/NotificationDispatcher';
+import { WebhooksDispatcher } from '../tv-modules/webhooks/WebhooksDispatcher';
+
+const dispatchers: Dispatcher[] = [
+    new NotificationDispatcher(),
+    new WebhooksDispatcher(),
+];
 
-/**
- * Register all event handlers (called on app init, before JobQueue starts)
- */
 export function registerAllEventHandlers() {
-    registerNotificationEventHandlers();
+    dispatchers.forEach((d) => d.register());
 }
 
-/**
- * Start JobQueue and register all job workers
- */
 export async function startAllWorkers() {
     await startJobQueue();
-    await registerNotificationWorkers();
+    for (const d of dispatchers) {
+        await d.registerWorkers();
+    }
 }

api/src/migrations/taskview/migrate.json

@@ -350,5 +350,27 @@
         "description": [
             "Added reminders, notifications, and push_subscriptions tables"
         ]
+    },
+    "28": {
+        "version": "1.25.0",
+        "name": "Release 1.25.0",
+        "releaseDate": "20260317",
+        "scripts": [
+            "/1.25.0/0.1.25.0.sql"
+        ],
+        "description": [
+            "Added type column to notifications table"
+        ]
+    },
+    "29": {
+        "version": "1.26.0",
+        "name": "Release 1.26.0",
+        "releaseDate": "20260320",
+        "scripts": [
+            "/1.26.0/0.1.26.0.sql"
+        ],
+        "description": [
+            "Added notification_preferences table with JSONB settings"
+        ]
     }
 }

api/src/migrations/taskview/sql/1.25.0/0.1.25.0.sql

@@ -0,0 +1,11 @@
+ALTER TABLE tasks.notifications
+    ADD COLUMN IF NOT EXISTS type VARCHAR(50) NOT NULL DEFAULT 'deadline';
+
+CREATE TABLE IF NOT EXISTS tasks.device_tokens (
+    id INTEGER GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
+    user_id INTEGER NOT NULL REFERENCES tv_auth.users(id) ON DELETE CASCADE,
+    token VARCHAR(500) NOT NULL,
+    platform VARCHAR(20) NOT NULL,
+    created_at TIMESTAMP NOT NULL DEFAULT NOW(),
+    CONSTRAINT uq_device_token UNIQUE (user_id, token)
+);

api/src/migrations/taskview/sql/1.25.0/1.migrate-time-to-utc.sql

@@ -0,0 +1,5 @@
+-- Convert TIMETZ columns to TIME (without timezone)
+-- Existing values are converted to UTC automatically by "AT TIME ZONE 'UTC'"
+ALTER TABLE tasks.tasks
+  ALTER COLUMN start_time TYPE TIME USING start_time AT TIME ZONE 'UTC',
+  ALTER COLUMN end_time TYPE TIME USING end_time AT TIME ZONE 'UTC';

api/src/migrations/taskview/sql/1.25.0/2.device-tokens-timezone.sql

@@ -0,0 +1,2 @@
+ALTER TABLE tasks.device_tokens
+  ADD COLUMN IF NOT EXISTS timezone VARCHAR(50) NOT NULL DEFAULT 'UTC';

api/src/migrations/taskview/sql/1.26.0/0.1.26.0.sql

@@ -0,0 +1,4 @@
+CREATE TABLE IF NOT EXISTS tasks.notification_preferences (
+    user_id INTEGER PRIMARY KEY REFERENCES tv_auth.users(id) ON DELETE CASCADE,
+    settings JSONB NOT NULL DEFAULT '{}'
+);

api/src/migrations/taskview/sql/1.27.0/0.1.27.0.sql

@@ -0,0 +1,26 @@
+CREATE TABLE IF NOT EXISTS tasks.webhooks (
+    id INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+    goal_id INTEGER NOT NULL REFERENCES tasks.goals(id) ON DELETE CASCADE,
+    url VARCHAR(500) NOT NULL,
+    secret_encrypted VARCHAR NOT NULL,
+    events VARCHAR[] NOT NULL DEFAULT '{}',
+    is_active BOOLEAN NOT NULL DEFAULT true,
+    consecutive_failures INTEGER NOT NULL DEFAULT 0,
+    created_at TIMESTAMP DEFAULT NOW(),
+    updated_at TIMESTAMP DEFAULT NOW()
+);
+
+CREATE TABLE IF NOT EXISTS tasks.webhook_deliveries (
+    id INTEGER PRIMARY KEY GENERATED ALWAYS AS IDENTITY,
+    webhook_id INTEGER NOT NULL REFERENCES tasks.webhooks(id) ON DELETE CASCADE,
+    event VARCHAR(50) NOT NULL,
+    payload JSONB NOT NULL,
+    status VARCHAR(20) NOT NULL DEFAULT 'pending',
+    response_code INTEGER,
+    attempts INTEGER NOT NULL DEFAULT 0,
+    last_attempt_at TIMESTAMP,
+    created_at TIMESTAMP DEFAULT NOW()
+);
+
+CREATE INDEX IF NOT EXISTS idx_webhooks_goal_id ON tasks.webhooks(goal_id);
+CREATE INDEX IF NOT EXISTS idx_webhook_deliveries_webhook_id ON tasks.webhook_deliveries(webhook_id);

api/src/routes/index.ts

@@ -5,6 +5,7 @@ import GoalsRoutes from '../tv-modules/goals/GoalsRoutes';
 import GraphRoutes from '../tv-modules/graph/GraphRoutes';
 import IntegrationsRoutes from '../tv-modules/integrations/IntegrationsRoutes';
 import NotificationsRoutes from '../tv-modules/notifications/NotificationsRoutes';
+import WebhooksRoutes from '../tv-modules/webhooks/WebhooksRoutes';
 import KanbanRoutes from '../tv-modules/kanban/KanbanRoutes';
 import GoalListRoutes from '../tv-modules/lists/GoalListRoutes';
 import StartRoutes from '../tv-modules/start/StartRoutes';
@@ -27,6 +28,7 @@ const routes: Record<string, RoutableConstructor> = {
     '/module/graph': GraphRoutes,
     '/module/integrations': IntegrationsRoutes,
     '/module/notifications': NotificationsRoutes,
+    '/module/webhooks': WebhooksRoutes,
 };
 
 export default routes;

api/src/tv-modules/kanban/KanbanController.ts

@@ -1,6 +1,7 @@
 import type { Request, Response } from 'express';
 import {
     KanbanArkTypeFetchTasksForColumn,
+    KanbanArkTypeFilters,
     KanbanArkTypeGetTasksOrderForColumnAndCursor,
     KanbanArkTypeUpdateTasksOrder,
     KanbanSchemaAddStatus,
@@ -53,7 +54,12 @@ export class KanbanController {
             return res.status(400).send(data.summary);
         }
 
-        return res.tvJson(await req.appUser.kanbanManager.fetchTasksForColumn(data));
+        const filters = KanbanArkTypeFilters(req.query);
+        if (filters instanceof ArkErrors) {
+            return res.status(400).send(filters.summary);
+        }
+
+        return res.tvJson(await req.appUser.kanbanManager.fetchTasksForColumn({ ...data, filters }));
     };
 
 

api/src/tv-modules/kanban/KanbanManager.ts

@@ -6,6 +6,7 @@ import {
     type DeleteKanbanStatus,
     type KanbanAddStatus,
     type KanbanArgFetchTasksForColumn,
+    type KanbanArgFilters,
     type KanbanArgGetTasksOrderForColumnAndCursor,
     type KanbanArgUpdateTasksOrder,
     type KanbanStatusForClient,
@@ -44,7 +45,7 @@ export class KanbanManager {
         return KanbanStatusToClientSchema.parse(status);
     }
 
-    async fetchTasksForColumn(data: KanbanArgFetchTasksForColumn): Promise<{ tasks: TaskForClientNew[], nextCursor: string | number | null, columnVersion: number | null }> {
+    async fetchTasksForColumn(data: KanbanArgFetchTasksForColumn & { filters?: KanbanArgFilters }): Promise<{ tasks: TaskForClientNew[], nextCursor: string | number | null, columnVersion: number | null }> {
         const [tasks, columnVersion] = await Promise.all([
             this.user.tasksManager.fetchTasksForKanbanColumn(data),
             this.repository.getColumnVersion(data.goalId, data.columnId)
@@ -176,8 +177,7 @@ export class KanbanManager {
                 }
             }
         } else {
-            // первая задача в колонке
-            newOrder = KANBAN_GAP;
+            newOrder = await this.user.tasksManager.repository.getNextKanbanOrder(data.goalId);
         }
 
         if (newOrder !== null) {

api/src/tv-modules/kanban/types.ts

@@ -54,14 +54,28 @@ export const KanbanArkTypeStatusToClient = type({
 
 export type KanbanStatusClient = typeof KanbanArkTypeStatusToClient.infer;
 
+const NullableNumberFromString = type('string|number|null').pipe((v) => (v === 'null' || v === null || v === undefined) ? null : Number(v));
+
 export const KanbanArkTypeFetchTasksForColumn = type({
     goalId: NumberFromString,
-    columnId: type('string|number|null').pipe((v) => (v === 'null' || v === null) ? null : Number(v)),
-    cursor: type('string|number|null').pipe((v) => (v === 'null' || v === null) ? null : Number(v)),
+    columnId: NullableNumberFromString,
+    cursor: NullableNumberFromString,
 });
 
 export type KanbanArgFetchTasksForColumn = typeof KanbanArkTypeFetchTasksForColumn.infer;
 
+const NumberArrayFromCommaSeparatedString = type('string|undefined').pipe((v) => {
+    if (!v) return [];
+    return v.split(',').map(Number).filter((n) => !isNaN(n));
+});
+
+export const KanbanArkTypeFilters = type({
+    'listIds?': NumberArrayFromCommaSeparatedString,
+    'assigneeIds?': NumberArrayFromCommaSeparatedString,
+});
+
+export type KanbanArgFilters = typeof KanbanArkTypeFilters.infer;
+
 export const KanbanArkTypeGetTasksOrderForColumnAndCursor = type({
     goalId: NumberFromString,
     columnId: type('string|number|null').pipe((v) => (v === 'null' || v === null) ? null : Number(v)),

api/src/tv-modules/notifications/NotificationDispatcher.ts

@@ -0,0 +1,164 @@
+import { eq, and, or, isNull, inArray } from 'drizzle-orm';
+import { alias } from 'drizzle-orm/pg-core';
+import { TasksSchema, CollaborationUsersSchema, UsersSchema } from 'taskview-db-schemas';
+import { eventBus, type AppEvents } from '../../core/EventBus';
+import { getJobQueue } from '../../core/JobQueue';
+import { Database } from '../../modules/db';
+import { $logger } from '../../modules/logget';
+import { getNotificationService } from './NotificationService';
+import { NotificationMessages } from './NotificationMessages';
+import { NotificationsRepository } from './repositories/NotificationsRepository';
+import { DeviceTokensRepository } from './repositories/DeviceTokensRepository';
+import { DeadlineScheduler } from './schedulers/DeadlineScheduler';
+import { NotificationType } from './types';
+import { parseUtcTime } from './utils';
+import type { Dispatcher } from '../../core/Dispatcher';
+
+const CLEANUP_JOB = 'notifications-cleanup';
+const NOTIFICATIONS_RETENTION_DAYS = 1;
+
+export class NotificationDispatcher implements Dispatcher {
+    private readonly deadlineScheduler = new DeadlineScheduler();
+    private readonly notificationsRepo = new NotificationsRepository();
+    private readonly deviceTokensRepo = new DeviceTokensRepository();
+
+    register(): void {
+        eventBus.on('task.created', (data) => this.onTaskCreated(data));
+        eventBus.on('task.updated', (data) => this.onTaskUpdated(data));
+        eventBus.on('task.assigneesChanged', (data) => this.onAssigneesChanged(data));
+        eventBus.on('task.deleted', (data) => this.onTaskDeleted(data));
+    }
+
+    async registerWorkers(): Promise<void> {
+        await this.cleanupWorker();
+        await this.deadlineScheduler.registerWorker();
+    }
+
+    private async cleanupWorker(): Promise<void> {
+        const boss = getJobQueue();
+        await boss.createQueue(CLEANUP_JOB);
+        await boss.schedule(CLEANUP_JOB, '0 3 * * *');
+        await boss.work(CLEANUP_JOB, async () => {
+            await this.notificationsRepo.deleteOlderThanDays(NOTIFICATIONS_RETENTION_DAYS);
+        });
+    }
+
+    private async onTaskCreated(data: AppEvents['task.created']): Promise<void> {
+        if (data.task.endDate) {
+            await this.deadlineScheduler.schedule(data.task, data.initiatorId);
+        }
+    }
+
+    private async onTaskUpdated(data: AppEvents['task.updated']): Promise<void> {
+        if (data.changes.complete === true) {
+            this.notificationsRepo.deleteByTaskAndType(data.task.id, NotificationType.DEADLINE);
+            await this.deadlineScheduler.cancel(data.task.id);
+            return;
+        }
+
+        const hasDeadlineChange =
+            data.changes.endDate !== undefined ||
+            data.changes.endTime !== undefined;
+
+        if (!hasDeadlineChange) return;
+
+        $logger.info(`[NotificationDispatcher] Rescheduling deadline for task=${data.task.id}`);
+
+        this.notificationsRepo.deleteByTaskAndType(data.task.id, NotificationType.DEADLINE);
+
+        if (data.task.endDate) {
+            await this.deadlineScheduler.schedule(data.task, data.initiatorId);
+        } else {
+            await this.deadlineScheduler.cancel(data.task.id);
+        }
+    }
+
+    private async onAssigneesChanged(data: AppEvents['task.assigneesChanged']): Promise<void> {
+        if (data.userIds.length === 0) return;
+
+        const db = Database.getInstance();
+        const task = await db.dbDrizzle
+            .select()
+            .from(TasksSchema)
+            .where(and(eq(TasksSchema.id, data.taskId), or(eq(TasksSchema.complete, false), isNull(TasksSchema.complete))));
+
+        if (!task[0]) return;
+
+        const authUsers = alias(UsersSchema, 'auth_users');
+        const authUserRows = await db.dbDrizzle
+            .select({ userId: authUsers.id })
+            .from(CollaborationUsersSchema)
+            .innerJoin(authUsers, eq(CollaborationUsersSchema.email, authUsers.email))
+            .where(inArray(CollaborationUsersSchema.id, data.userIds));
+
+        const recipientIds = authUserRows
+            .map((r) => r.userId)
+            .filter((id) => id !== data.initiatorId);
+
+        if (recipientIds.length === 0) return;
+
+        await this.handleAssignNotification(data, task[0], recipientIds);
+        await this.handleExpiredDeadlineNotification(data, task[0], recipientIds);
+    }
+
+    private async handleAssignNotification(
+        data: AppEvents['task.assigneesChanged'],
+        task: typeof TasksSchema.$inferSelect,
+        recipientIds: number[],
+    ): Promise<void> {
+        const initiatorName = await this.resolveUserName(data.initiatorId);
+        const message = NotificationMessages.assign(task.description, initiatorName);
+
+        $logger.info(`[NotificationDispatcher] Assign notification for task=${data.taskId}, recipients=[${recipientIds.join(',')}]`);
+
+        await getNotificationService().notifyMany(
+            recipientIds,
+            NotificationType.ASSIGN,
+            message,
+            { goalId: task.goalId, goalListId: task.goalListId },
+            data.taskId,
+        );
+    }
+
+    private async handleExpiredDeadlineNotification(
+        data: AppEvents['task.assigneesChanged'],
+        task: typeof TasksSchema.$inferSelect,
+        recipientIds: number[],
+    ): Promise<void> {
+        if (!task.endDate) return;
+
+        const isExpired = task.endTime
+            ? (parseUtcTime(task.endDate, task.endTime) ?? new Date()) <= new Date()
+            : new Date(`${task.endDate}T00:00:00Z`) <= new Date();
+
+        if (!isExpired) return;
+
+        const tz = task.owner ? await this.deviceTokensRepo.getTimezoneByUserId(task.owner) : 'UTC';
+        const message = NotificationMessages.deadline(task.description, task.endDate, task.endTime, tz);
+
+        $logger.info(`[NotificationDispatcher] Expired deadline notification for task=${data.taskId}, recipients=[${recipientIds.join(',')}]`);
+
+        await getNotificationService().notifyMany(
+            recipientIds,
+            NotificationType.DEADLINE,
+            message,
+            { goalId: task.goalId, goalListId: task.goalListId },
+            data.taskId,
+        );
+    }
+
+    private async onTaskDeleted(data: AppEvents['task.deleted']): Promise<void> {
+        this.notificationsRepo.deleteByTaskAndType(data.taskId, NotificationType.DEADLINE);
+        await this.deadlineScheduler.cancel(data.taskId);
+    }
+
+    private async resolveUserName(userId: number): Promise<string> {
+        const db = Database.getInstance();
+        const result = await db.dbDrizzle
+            .select({ login: UsersSchema.login })
+            .from(UsersSchema)
+            .where(eq(UsersSchema.id, userId))
+            .limit(1);
+        return result[0]?.login || 'Someone';
+    }
+}

api/src/tv-modules/notifications/NotificationMessages.ts

@@ -0,0 +1,51 @@
+import type { NotificationMessage } from './types';
+import { parseUtcTime } from './utils';
+
+export class NotificationMessages {
+    static deadline(description: string | null, endDate: string, endTime: string | null, timezone: string): NotificationMessage {
+        const title = `Task: ${description || 'Task'}`;
+
+        if (endTime) {
+            const deadline = parseUtcTime(endDate, endTime);
+            if (deadline) {
+                const formatted = deadline.toLocaleString('en-US', {
+                    timeZone: timezone,
+                    month: 'short', day: 'numeric',
+                    hour: '2-digit', minute: '2-digit',
+                    hour12: false,
+                });
+                return { title, body: `Deadline: ${formatted}` };
+            }
+        }
+
+        return { title, body: `Deadline: ${endDate}` };
+    }
+
+    static assign(taskDescription: string | null, assignedByName: string): NotificationMessage {
+        return {
+            title: `Task: ${taskDescription || 'Task'}`,
+            body: `Assigned to you by ${assignedByName}`,
+        };
+    }
+
+    static mention(taskDescription: string | null, mentionedByName: string): NotificationMessage {
+        return {
+            title: `Task: ${taskDescription || 'Task'}`,
+            body: `${mentionedByName} mentioned you`,
+        };
+    }
+
+    static comment(taskDescription: string | null, commentByName: string): NotificationMessage {
+        return {
+            title: `Task: ${taskDescription || 'Task'}`,
+            body: `New comment by ${commentByName}`,
+        };
+    }
+
+    static statusChange(taskDescription: string | null, newStatus: string): NotificationMessage {
+        return {
+            title: `Task: ${taskDescription || 'Task'}`,
+            body: `Status changed to ${newStatus}`,
+        };
+    }
+}

api/src/tv-modules/notifications/NotificationProvider.ts

@@ -0,0 +1,12 @@
+import type { NotificationsSchemaTypeForSelect } from 'taskview-db-schemas';
+import type { NotificationChannel } from './types';
+
+export interface NotificationMeta {
+    goalId: number;
+    goalListId: number | null;
+}
+
+export interface NotificationProvider {
+    readonly channel: NotificationChannel;
+    send(userId: number, notification: NotificationsSchemaTypeForSelect, meta: NotificationMeta): Promise<void>;
+}

api/src/tv-modules/notifications/NotificationService.ts

@@ -0,0 +1,85 @@
+import { $logger } from '../../modules/logget';
+import type { NotificationMeta, NotificationProvider } from './NotificationProvider';
+import type { NotificationMessage } from './types';
+import { NotificationsRepository } from './repositories/NotificationsRepository';
+import { UserPreferencesRepository } from './repositories/UserPreferencesRepository';
+import { NotificationChannel, type NotificationType } from './types';
+import { CentrifugoProvider } from './providers/CentrifugoProvider';
+import { FCMProvider } from './providers/FCMProvider';
+
+export class NotificationService {
+    private readonly providers: Map<NotificationChannel, NotificationProvider>;
+    private readonly repo: NotificationsRepository;
+    private readonly preferences: UserPreferencesRepository;
+
+    constructor() {
+        this.repo = new NotificationsRepository();
+        this.preferences = new UserPreferencesRepository();
+
+        const providerList: NotificationProvider[] = [
+            new CentrifugoProvider(),
+            new FCMProvider(),
+        ];
+
+        this.providers = new Map();
+        for (const p of providerList) {
+            this.providers.set(p.channel, p);
+        }
+    }
+
+    async notify(
+        userId: number,
+        type: NotificationType,
+        message: NotificationMessage,
+        meta: NotificationMeta,
+        taskId: number | null = null,
+    ): Promise<void> {
+        $logger.info(`[NotificationService] notify user=${userId}, type=${type}, title="${message.title}"`);
+
+        const channels = await this.preferences.getEnabledChannels(userId, type, meta.goalId);
+        if (channels.length === 0) {
+            $logger.info(`[NotificationService] All channels disabled for user=${userId}, type=${type}`);
+            return;
+        }
+
+        const notification = await this.repo.create({ userId, taskId, type, title: message.title, body: message.body });
+        if (!notification) {
+            $logger.warn(`[NotificationService] Failed to create notification record for user ${userId}`);
+            return;
+        }
+
+        $logger.info(`[NotificationService] Created notification id=${notification.id}, sending to channels=[${channels.join(',')}]`);
+
+        await Promise.allSettled(
+            channels
+                .map((channel) => this.providers.get(channel))
+                .filter(Boolean)
+                .map((provider) =>
+                    provider!.send(userId, notification, meta).catch((err) => {
+                        $logger.error(err, `[NotificationService] Provider "${provider!.channel}" failed for user ${userId}`);
+                    })
+                )
+        );
+    }
+
+    async notifyMany(
+        userIds: number[],
+        type: NotificationType,
+        message: NotificationMessage,
+        meta: NotificationMeta,
+        taskId: number | null = null,
+    ): Promise<void> {
+        await Promise.allSettled(
+            userIds.map((userId) => this.notify(userId, type, message, meta, taskId))
+        );
+    }
+}
+
+let _instance: NotificationService | null = null;
+
+export function getNotificationService(): NotificationService {
+    if (!_instance) {
+        _instance = new NotificationService();
+    }
+    return _instance;
+}

api/src/tv-modules/notifications/NotificationsController.ts

@@ -1,11 +1,20 @@
 import { type } from 'arktype';
 import type { Request, Response } from 'express';
 import { CentrifugoClient } from '../../core/CentrifugoClient';
+import { DeviceTokensRepository } from './repositories/DeviceTokensRepository';
+import { UserPreferencesRepository } from './repositories/UserPreferencesRepository';
+import { UserPreferences } from './UserPreferences';
 
 const NotificationArkTypeMarkRead = type({
     notificationId: 'number',
 });
 
+const DeviceTokenArkType = type({
+    token: 'string',
+    platform: "'android' | 'ios'",
+    timezone: 'string',
+});
+
 export class NotificationsController {
     fetch = async (req: Request, res: Response) => {
         const cursor = req.query.cursor ? Number(req.query.cursor) : undefined;
@@ -24,6 +33,56 @@ export class NotificationsController {
         return res.tvJson(await _req.appUser.notificationsManager.markAllRead());
     };
 
+    registerDevice = async (req: Request, res: Response) => {
+        console.log('[registerDevice] body:', JSON.stringify(req.body));
+        const out = DeviceTokenArkType(req.body);
+        if (out instanceof type.errors) {
+            console.log('[registerDevice] validation error:', out.summary);
+            return res.status(400).send(out.summary);
+        }
+        const userId = req.appUser.getUserData()?.id;
+        if (!userId) return res.status(401).send('Unauthorized');
+
+        const repo = new DeviceTokensRepository();
+        return res.tvJson(await repo.register(userId, out.token, out.platform, out.timezone));
+    };
+
+    unregisterDevice = async (req: Request, res: Response) => {
+        const { token } = req.body;
+        if (!token || typeof token !== 'string') {
+            return res.status(400).send('token is required');
+        }
+        const userId = req.appUser.getUserData()?.id;
+        if (!userId) return res.status(401).send('Unauthorized');
+
+        const repo = new DeviceTokensRepository();
+        return res.tvJson(await repo.unregister(userId, token));
+    };
+
+    getPreferences = async (req: Request, res: Response) => {
+        const userId = req.appUser.getUserData()?.id;
+        if (!userId) return res.status(400).send('Bad request');
+
+        const repo = new UserPreferencesRepository();
+        const prefs = await repo.load(userId);
+        return res.tvJson({ settings: prefs.toJSON() });
+    };
+
+    savePreferences = async (req: Request, res: Response) => {
+        const userId = req.appUser.getUserData()?.id;
+        if (!userId) return res.status(400).send('Bad request');
+
+        const { settings } = req.body;
+        if (!settings || typeof settings !== 'object') {
+            return res.status(400).send('settings is required');
+        }
+
+        const repo = new UserPreferencesRepository();
+        const prefs = new UserPreferences(settings);
+        const result = await repo.save(userId, prefs);
+        return res.tvJson(result);
+    };
+
     connectionToken = async (req: Request, res: Response) => {
         const userId = req.appUser.getUserData()?.id;
         if (!userId) {

api/src/tv-modules/notifications/NotificationsManager.ts

@@ -1,5 +1,5 @@
 import type { AppUser } from '../../core/AppUser';
-import { NotificationsRepository } from './NotificationsRepository';
+import { NotificationsRepository } from './repositories/NotificationsRepository';
 
 export class NotificationsManager {
     private readonly user: AppUser;

api/src/tv-modules/notifications/NotificationsRoutes.ts

@@ -21,6 +21,10 @@ export default class NotificationsRoutes implements Routable {
         this.router.get('/', [IsLoggedIn], this.controller.fetch);
         this.router.patch('/read', [IsLoggedIn], this.controller.markRead);
         this.router.patch('/read-all', [IsLoggedIn], this.controller.markAllRead);
+        this.router.get('/preferences', [IsLoggedIn], this.controller.getPreferences);
+        this.router.put('/preferences', [IsLoggedIn], this.controller.savePreferences);
         this.router.get('/connection-token', [IsLoggedIn], this.controller.connectionToken);
+        this.router.post('/device/register', [IsLoggedIn], this.controller.registerDevice);
+        this.router.post('/device/unregister', [IsLoggedIn], this.controller.unregisterDevice);
     }
 }

api/src/tv-modules/notifications/UserPreferences.ts

@@ -0,0 +1,124 @@
+import {
+    NotificationChannel,
+    NotificationType,
+    type TypeSettings,
+    type TypeSettingsMap,
+    type DeadlineTypeSettings,
+    type DeadlineIntervals,
+    type SettingsJson,
+} from './types';
+
+const ALL_CHANNELS: NotificationChannel[] = [
+    NotificationChannel.PUSH,
+    NotificationChannel.WEBSOCKET
+];
+
+/**
+ * Manages user notification preferences.
+ * Opt-out model: undefined = enabled.
+ *
+ * Priority:
+ *   1. projects[goalId][type] — highest
+ *   2. global[type] — fallback
+ *   3. undefined — enabled
+ */
+export class UserPreferences {
+    private data: SettingsJson;
+
+    constructor(json: unknown) {
+        this.data = (json && typeof json === 'object' ? json : {}) as SettingsJson;
+    }
+
+    getEnabledChannels(type: NotificationType, goalId?: number): NotificationChannel[] {
+        const resolved = this.resolveTypeSettings(type, goalId);
+        if (!resolved?.channels) return ALL_CHANNELS;
+        return ALL_CHANNELS.filter((ch) => resolved.channels![ch] !== false);
+    }
+
+    isChannelEnabled(type: NotificationType, channel: NotificationChannel, goalId?: number): boolean {
+        const resolved = this.resolveTypeSettings(type, goalId);
+        return resolved?.channels?.[channel] !== false;
+    }
+
+    getDeadlineIntervals(goalId?: number): DeadlineIntervals | undefined {
+        const resolved = this.resolveTypeSettings(NotificationType.DEADLINE, goalId) as DeadlineTypeSettings | undefined;
+        return resolved?.intervals;
+    }
+
+    getEnabledDeadlineIntervals(goalId?: number): number[] {
+        const intervals = this.getDeadlineIntervals(goalId);
+        if (!intervals) return [0]; // default: notify at deadline
+        return Object.entries(intervals)
+            .filter(([, enabled]) => enabled !== false)
+            .map(([minutes]) => Number(minutes));
+    }
+
+    getGlobalTypeSettings<T extends NotificationType>(type: T): TypeSettingsMap[T] | undefined {
+        return this.data.global?.[type] as TypeSettingsMap[T] | undefined;
+    }
+
+    getProjectTypeSettings<T extends NotificationType>(goalId: number, type: T): TypeSettingsMap[T] | undefined {
+        return this.data.projects?.[String(goalId)]?.[type] as TypeSettingsMap[T] | undefined;
+    }
+
+    setGlobalChannel(type: NotificationType, channel: NotificationChannel, enabled: boolean): void {
+        if (!this.data.global) this.data.global = {};
+        if (!this.data.global[type]) this.data.global[type] = {} as TypeSettingsMap[typeof type];
+        if (!this.data.global[type]!.channels) this.data.global[type]!.channels = {};
+        this.data.global[type]!.channels![channel] = enabled;
+    }
+
+    setDeadlineIntervals(intervals: DeadlineIntervals, goalId?: number): void {
+        if (goalId !== undefined) {
+            const key = String(goalId);
+            if (!this.data.projects) this.data.projects = {};
+            if (!this.data.projects[key]) this.data.projects[key] = {};
+            if (!this.data.projects[key][NotificationType.DEADLINE]) this.data.projects[key][NotificationType.DEADLINE] = {};
+            (this.data.projects[key][NotificationType.DEADLINE] as DeadlineTypeSettings).intervals = intervals;
+        } else {
+            if (!this.data.global) this.data.global = {};
+            if (!this.data.global[NotificationType.DEADLINE]) this.data.global[NotificationType.DEADLINE] = {};
+            (this.data.global[NotificationType.DEADLINE] as DeadlineTypeSettings).intervals = intervals;
+        }
+    }
+
+    setProjectChannel(goalId: number, type: NotificationType, channel: NotificationChannel, enabled: boolean): void {
+        const key = String(goalId);
+        if (!this.data.projects) this.data.projects = {};
+        if (!this.data.projects[key]) this.data.projects[key] = {};
+        if (!this.data.projects[key][type]) this.data.projects[key][type] = {} as TypeSettingsMap[typeof type];
+        if (!this.data.projects[key][type]!.channels) this.data.projects[key][type]!.channels = {};
+        this.data.projects[key][type]!.channels![channel] = enabled;
+    }
+
+    removeProjectSettings(goalId: number): void {
+        delete this.data.projects?.[String(goalId)];
+    }
+
+    toJSON(): SettingsJson {
+        return this.data;
+    }
+
+    private resolveTypeSettings(type: NotificationType, goalId?: number): TypeSettings | undefined {
+        const globalSettings = this.data.global?.[type];
+
+        if (goalId === undefined) return globalSettings;
+
+        const projectSettings = this.data.projects?.[String(goalId)]?.[type];
+        if (!projectSettings) return globalSettings;
+        if (!globalSettings) return projectSettings;
+
+        return {
+            channels: { ...globalSettings.channels, ...projectSettings.channels },
+            ...('intervals' in globalSettings || 'intervals' in projectSettings
+                ? {
+                    intervals: {
+                        ...(globalSettings as DeadlineTypeSettings).intervals,
+                        ...(projectSettings as DeadlineTypeSettings).intervals,
+                    }
+                }
+                : {}
+            ),
+        };
+    }
+}

api/src/tv-modules/notifications/notifications.events.ts

@@ -1,181 +0,0 @@
-import { eventBus } from '../../core/EventBus';
-import { getJobQueue } from '../../core/JobQueue';
-import { NotificationsRepository } from './NotificationsRepository';
-import { getCentrifugoClient } from '../../core/CentrifugoClient';
-import { $logger } from '../../modules/logget';
-import { Database } from '../../modules/db';
-import { eq, and, or, isNull } from 'drizzle-orm';
-import { alias } from 'drizzle-orm/pg-core';
-import { TasksSchema, TasksAssigneeSchema, GoalsSchema, CollaborationUsersSchema, UsersSchema } from 'taskview-db-schemas';
-
-const DEADLINE_JOB = 'deadline-notification';
-const CLEANUP_JOB = 'notifications-cleanup';
-const NOTIFICATIONS_RETENTION_DAYS = 1;
-
-interface DeadlineJobData {
-    taskId: number;
-    description: string;
-    goalId: number;
-    goalListId: number | null;
-    owner: number;
-    endDate: string;
-    endTime: string | null;
-    initiatorId: number | null;
-}
-
-async function scheduleDeadlineJob(task: { id: number; description: string | null; goalId: number; goalListId: number | null; owner: number | null; endDate: string | null; endTime: string | null }, initiatorId?: number) {
-    if (!task.endDate) return;
-
-    const boss = getJobQueue();
-    const jobKey = `deadline-${task.id}`;
-    const startAfter = new Date(task.endDate);
-
-    if (task.endTime) {
-        const [h, m] = task.endTime.split(':');
-        startAfter.setHours(Number(h), Number(m), 0, 0);
-    }
-
-    const now = new Date();
-    await boss.send(DEADLINE_JOB, {
-        taskId: task.id,
-        description: task.description,
-        goalId: task.goalId,
-        goalListId: task.goalListId,
-        owner: task.owner,
-        endDate: task.endDate,
-        endTime: task.endTime,
-        initiatorId: initiatorId ?? null,
-    }, {
-        startAfter: startAfter >= now ? startAfter : undefined,
-        singletonKey: jobKey,
-    });
-    $logger.info({ taskId: task.id, startAfter }, '[Notifications] Deadline job scheduled');
-}
-
-export function registerNotificationEventHandlers() {
-    const repo = new NotificationsRepository();
-
-    // New task created with deadline
-    eventBus.on('task.created', async (data) => {
-        if (data.task.endDate) {
-            await scheduleDeadlineJob(data.task, data.userId);
-        }
-    });
-
-    // Deadline changed on existing task
-    eventBus.on('task.updated', async (data) => {
-        const hasDeadlineChange =
-            data.changes.endDate !== undefined ||
-            data.changes.startDate !== undefined ||
-            data.changes.endTime !== undefined ||
-            data.changes.startTime !== undefined;
-
-        if (!hasDeadlineChange) return;
-
-        repo.deleteDeadlineNotifications(data.task.id);
-        await scheduleDeadlineJob(data.task, data.userId);
-    });
-
-    // Assignees changed — reschedule deadline job so new users get notified
-    eventBus.on('task.assigneesChanged', async (data) => {
-        const db = Database.getInstance();
-        const task = await db.dbDrizzle
-            .select()
-            .from(TasksSchema)
-            .where(eq(TasksSchema.id, data.taskId));
-
-        if (task[0]?.endDate) {
-            await scheduleDeadlineJob(task[0]);
-        }
-    });
-}
-
-export async function registerNotificationWorkers() {
-    const boss = getJobQueue();
-    const repo = new NotificationsRepository();
-    const db = Database.getInstance();
-
-    await boss.createQueue(DEADLINE_JOB);
-    await boss.createQueue(CLEANUP_JOB);
-
-    // Worker: process deadline notifications
-    await boss.work<DeadlineJobData>(DEADLINE_JOB, async ([job]) => {
-        $logger.info({ jobData: job.data }, '[Notifications] Deadline worker received job');
-        const { taskId, description, goalId, goalListId, endDate, endTime, initiatorId } = job.data;
-
-        // Check if task is still active (not completed or deleted)
-        const task = await db.dbDrizzle
-            .select({ owner: TasksSchema.owner })
-            .from(TasksSchema)
-            .where(and(eq(TasksSchema.id, taskId), or(eq(TasksSchema.complete, false), isNull(TasksSchema.complete))));
-
-        $logger.info({ taskId, taskResult: task }, '[Notifications] Task lookup result');
-        if (task.length === 0) return;
-
-        const taskOwner = task[0].owner;
-
-        // Collect all recipients: task owner + assignees (resolved to auth user id) + goal owner
-        let assigneesResult: { userId: number }[] = [];
-        let goalResult: { owner: number }[] = [];
-        const authUsers = alias(UsersSchema, 'auth_users');
-        try {
-        [assigneesResult, goalResult] = await Promise.all([
-            db.dbDrizzle
-                .select({ userId: authUsers.id })
-                .from(TasksAssigneeSchema)
-                .innerJoin(CollaborationUsersSchema, eq(TasksAssigneeSchema.collabUserId, CollaborationUsersSchema.id))
-                .innerJoin(authUsers, eq(CollaborationUsersSchema.email, authUsers.email))
-                .where(eq(TasksAssigneeSchema.taskId, taskId)),
-            db.dbDrizzle
-                .select({ owner: GoalsSchema.owner })
-                .from(GoalsSchema)
-                .where(eq(GoalsSchema.id, goalId)),
-        ]);
-        } catch (err) {
-            $logger.error(err, '[Notifications] Failed to resolve recipients');
-        }
-
-        const recipientIds = new Set<number>();
-        if (taskOwner) recipientIds.add(taskOwner);
-        assigneesResult.forEach((r) => recipientIds.add(r.userId));
-        if (goalResult[0]) {
-            recipientIds.add(goalResult[0].owner);
-        }
-
-        // Don't notify the person who made the change
-        if (initiatorId) recipientIds.delete(initiatorId);
-
-        $logger.info({ taskId, recipientIds: [...recipientIds], assignees: assigneesResult, goalOwner: goalResult[0]?.owner }, '[Notifications] Recipients resolved');
-
-        const title = `Deadline: ${description || 'Task'}`;
-        const timeStr = endTime ? `${endDate} ${endTime}` : endDate;
-        const body = `Deadline approaching: ${timeStr}`;
-
-        for (const userId of recipientIds) {
-            const notification = await repo.create({
-                userId,
-                taskId,
-                title,
-                body,
-            });
-
-            if (notification) {
-                await getCentrifugoClient().publishToUser(userId, 'notification', {
-                    notification,
-                    goalId,
-                    goalListId,
-                });
-            }
-        }
-
-        $logger.info({ taskId, recipients: [...recipientIds] }, '[Notifications] Deadline notifications sent');
-    });
-
-    // Schedule periodic cleanup of old notifications
-    await boss.schedule(CLEANUP_JOB, '0 3 * * *'); // daily at 3:00 AM
-
-    await boss.work(CLEANUP_JOB, async (_jobs) => {
-        await repo.deleteOlderThanDays(NOTIFICATIONS_RETENTION_DAYS);
-        $logger.info('[Notifications] Old notifications cleaned up');
-    });
-}

api/src/tv-modules/notifications/providers/CentrifugoProvider.ts

@@ -0,0 +1,16 @@
+import type { NotificationsSchemaTypeForSelect } from 'taskview-db-schemas';
+import { getCentrifugoClient } from '../../../core/CentrifugoClient';
+import type { NotificationMeta, NotificationProvider } from '../NotificationProvider';
+import { NotificationChannel } from '../types';
+
+export class CentrifugoProvider implements NotificationProvider {
+    readonly channel = NotificationChannel.WEBSOCKET;
+
+    async send(userId: number, notification: NotificationsSchemaTypeForSelect, meta: NotificationMeta): Promise<void> {
+        await getCentrifugoClient().publishToUser(userId, 'notification', {
+            notification,
+            goalId: meta.goalId,
+            goalListId: meta.goalListId,
+        });
+    }
+}

api/src/tv-modules/notifications/providers/FCMProvider.ts

@@ -0,0 +1,107 @@
+import admin from 'firebase-admin';
+import { getMessaging } from 'firebase-admin/messaging';
+import type { NotificationsSchemaTypeForSelect } from 'taskview-db-schemas';
+import type { NotificationMeta, NotificationProvider } from '../NotificationProvider';
+import { NotificationChannel } from '../types';
+import { DeviceTokensRepository } from '../repositories/DeviceTokensRepository';
+import { $logger } from '../../../modules/logget';
+
+export class FCMProvider implements NotificationProvider {
+    readonly channel = NotificationChannel.PUSH;
+
+    private static initialized = false;
+    private static messaging: admin.messaging.Messaging | null = null;
+
+    private readonly repo = new DeviceTokensRepository();
+    private readonly enabled: boolean;
+
+    constructor() {
+        this.enabled = FCMProvider.init();
+    }
+
+    private static init(): boolean {
+        if (FCMProvider.initialized) return true;
+
+        const credentialsPath = process.env.FIREBASE_CREDENTIALS_PATH;
+        if (!credentialsPath) {
+            $logger.warn('[FCM] FIREBASE_CREDENTIALS_PATH not configured — push notifications disabled');
+            return false;
+        }
+
+        try {
+            admin.initializeApp({
+                credential: admin.credential.cert(credentialsPath),
+            });
+            FCMProvider.messaging = getMessaging();
+            FCMProvider.messaging.enableLegacyHttpTransport();
+            FCMProvider.initialized = true;
+            $logger.info('[FCM] Firebase initialized with legacy HTTP/1.1 transport');
+            return true;
+        } catch (err) {
+            $logger.error(err, '[FCM] Failed to initialize Firebase');
+            return false;
+        }
+    }
+
+    async send(userId: number, notification: NotificationsSchemaTypeForSelect, meta: NotificationMeta): Promise<void> {
+        if (!this.enabled || !FCMProvider.messaging) return;
+
+        const tokens = await this.repo.getByUserId(userId);
+        $logger.info(`[FCM] User ${userId}: found ${tokens.length} device token(s)`);
+        if (tokens.length === 0) return;
+
+        const message: admin.messaging.MulticastMessage = {
+            tokens: tokens.map((t) => t.token),
+            notification: {
+                title: notification.title,
+                body: notification.body || undefined,
+            },
+            data: {
+                type: notification.type,
+                taskId: notification.taskId ? String(notification.taskId) : '',
+                goalId: String(meta.goalId),
+                goalListId: meta.goalListId ? String(meta.goalListId) : '',
+                notificationId: String(notification.id),
+            },
+            android: {
+                priority: 'high',
+                notification: {
+                    sound: 'default',
+                },
+            },
+            apns: {
+                payload: {
+                    aps: {
+                        sound: 'default',
+                    },
+                },
+            },
+        };
+
+        try {
+            $logger.info(`[FCM] Sending to ${tokens.length} token(s) for user ${userId}, title="${notification.title}"`);
+            const response = await FCMProvider.messaging.sendEachForMulticast(message);
+            $logger.info(`[FCM] Result: success=${response.successCount}, failure=${response.failureCount}`);
+
+            if (response.failureCount > 0) {
+                const invalidTokens: string[] = [];
+                response.responses.forEach((resp, idx) => {
+                    if (!resp.success) {
+                        const code = resp.error?.code;
+                        if (code === 'messaging/invalid-registration-token' || code === 'messaging/registration-token-not-registered') {
+                            invalidTokens.push(tokens[idx].token);
+                        } else {
+                            $logger.error(resp.error, '[FCM] Failed to send to token');
+                        }
+                    }
+                });
+
+                for (const token of invalidTokens) {
+                    await this.repo.deleteByToken(token);
+                }
+            }
+        } catch (err) {
+            $logger.error(err, `[FCM] Failed to send multicast for user ${userId}`);
+        }
+    }
+}

api/src/tv-modules/notifications/repositories/DeviceTokensRepository.ts

@@ -0,0 +1,64 @@
+import { and, eq } from 'drizzle-orm';
+import { DeviceTokensSchema, type DeviceTokensSchemaTypeForSelect } from 'taskview-db-schemas';
+import { Database } from '../../../modules/db';
+import { callWithCatch } from '../../../utils/helpers';
+
+export class DeviceTokensRepository {
+    private readonly db: Database;
+
+    constructor() {
+        this.db = Database.getInstance();
+    }
+
+    async register(userId: number, token: string, platform: string, timezone: string): Promise<boolean> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.insert(DeviceTokensSchema).values({
+                userId,
+                token,
+                platform,
+                timezone,
+            }).onConflictDoUpdate({
+                target: [DeviceTokensSchema.userId, DeviceTokensSchema.token],
+                set: { timezone },
+            })
+        );
+        return !!result;
+    }
+
+    async unregister(userId: number, token: string): Promise<boolean> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.delete(DeviceTokensSchema)
+                .where(and(
+                    eq(DeviceTokensSchema.userId, userId),
+                    eq(DeviceTokensSchema.token, token),
+                ))
+        );
+        return !!result;
+    }
+
+    async getByUserId(userId: number): Promise<DeviceTokensSchemaTypeForSelect[]> {
+        return await callWithCatch(() =>
+            this.db.dbDrizzle.select()
+                .from(DeviceTokensSchema)
+                .where(eq(DeviceTokensSchema.userId, userId))
+        ) || [];
+    }
+
+    async getTimezoneByUserId(userId: number): Promise<string> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.select({ timezone: DeviceTokensSchema.timezone })
+                .from(DeviceTokensSchema)
+                .where(eq(DeviceTokensSchema.userId, userId))
+                .limit(1)
+        );
+        return result?.[0]?.timezone || 'UTC';
+    }
+
+    async deleteByToken(token: string): Promise<boolean> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.delete(DeviceTokensSchema)
+                .where(eq(DeviceTokensSchema.token, token))
+        );
+        return !!result;
+    }
+}

api/src/tv-modules/notifications/repositories/NotificationsRepository.ts

@@ -1,7 +1,7 @@
-import { and, eq, desc, lt, like, sql } from 'drizzle-orm';
+import { and, eq, desc, lt, sql } from 'drizzle-orm';
 import { NotificationsSchema, TasksSchema, type NotificationsSchemaTypeForSelect } from 'taskview-db-schemas';
-import { Database } from '../../modules/db';
-import { callWithCatch } from '../../utils/helpers';
+import { Database } from '../../../modules/db';
+import { callWithCatch } from '../../../utils/helpers';
 
 export class NotificationsRepository {
     private readonly db: Database;
@@ -10,11 +10,12 @@ export class NotificationsRepository {
         this.db = Database.getInstance();
     }
 
-    async create(data: { userId: number; taskId: number | null; title: string; body: string | null }): Promise<NotificationsSchemaTypeForSelect | false> {
+    async create(data: { userId: number; taskId: number | null; type: string; title: string; body: string | null }): Promise<NotificationsSchemaTypeForSelect | false> {
         const result = await callWithCatch(() =>
             this.db.dbDrizzle.insert(NotificationsSchema).values({
                 userId: data.userId,
                 taskId: data.taskId,
+                type: data.type,
                 title: data.title,
                 body: data.body,
             }).returning()
@@ -34,6 +35,7 @@ export class NotificationsRepository {
                 id: NotificationsSchema.id,
                 userId: NotificationsSchema.userId,
                 taskId: NotificationsSchema.taskId,
+                type: NotificationsSchema.type,
                 title: NotificationsSchema.title,
                 body: NotificationsSchema.body,
                 read: NotificationsSchema.read,
@@ -62,12 +64,12 @@ export class NotificationsRepository {
         return !!result;
     }
 
-    async deleteDeadlineNotifications(taskId: number): Promise<boolean> {
+    async deleteByTaskAndType(taskId: number, type: string): Promise<boolean> {
         const result = await callWithCatch(() =>
             this.db.dbDrizzle.delete(NotificationsSchema)
                 .where(and(
                     eq(NotificationsSchema.taskId, taskId),
-                    like(NotificationsSchema.title, 'Deadline:%'),
+                    eq(NotificationsSchema.type, type),
                 ))
         );
         return !!result;

api/src/tv-modules/notifications/repositories/UserPreferencesRepository.ts

@@ -0,0 +1,42 @@
+import { eq } from 'drizzle-orm';
+import { NotificationPreferencesSchema } from 'taskview-db-schemas';
+import { Database } from '../../../modules/db';
+import { callWithCatch } from '../../../utils/helpers';
+import { UserPreferences } from '../UserPreferences';
+import type { NotificationChannel, NotificationType } from '../types';
+
+export class UserPreferencesRepository {
+    private readonly db: Database;
+
+    constructor() {
+        this.db = Database.getInstance();
+    }
+
+    async load(userId: number): Promise<UserPreferences> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.select({ settings: NotificationPreferencesSchema.settings })
+                .from(NotificationPreferencesSchema)
+                .where(eq(NotificationPreferencesSchema.userId, userId))
+                .limit(1)
+        );
+        return new UserPreferences(result?.[0]?.settings);
+    }
+
+    async save(userId: number, preferences: UserPreferences): Promise<boolean> {
+        const settings = preferences.toJSON();
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.insert(NotificationPreferencesSchema)
+                .values({ userId, settings })
+                .onConflictDoUpdate({
+                    target: [NotificationPreferencesSchema.userId],
+                    set: { settings },
+                })
+        );
+        return !!result;
+    }
+
+    async getEnabledChannels(userId: number, type: NotificationType, goalId?: number): Promise<NotificationChannel[]> {
+        const prefs = await this.load(userId);
+        return prefs.getEnabledChannels(type, goalId);
+    }
+}

api/src/tv-modules/notifications/schedulers/DeadlineScheduler.ts

@@ -0,0 +1,142 @@
+import { eq, and, or, isNull } from 'drizzle-orm';
+import { alias } from 'drizzle-orm/pg-core';
+import { TasksSchema, TasksAssigneeSchema, GoalsSchema, CollaborationUsersSchema, UsersSchema } from 'taskview-db-schemas';
+import { getJobQueue, cancelJobBySingletonKey } from '../../../core/JobQueue';
+import { Database } from '../../../modules/db';
+import { $logger } from '../../../modules/logget';
+import { getNotificationService } from '../NotificationService';
+import { NotificationMessages } from '../NotificationMessages';
+import { DeviceTokensRepository } from '../repositories/DeviceTokensRepository';
+import { NotificationType, type DeadlineJobData, type TaskWithDeadline } from '../types';
+import { parseUtcTime } from '../utils';
+
+const DEADLINE_JOB = 'deadline-notification';
+
+export class DeadlineScheduler {
+    private readonly deviceTokensRepo = new DeviceTokensRepository();
+
+    async schedule(task: TaskWithDeadline, initiatorId?: number): Promise<void> {
+        if (!task.endDate) return;
+
+        let startAfter: Date | undefined;
+
+        if (task.endTime) {
+            const deadline = parseUtcTime(task.endDate, task.endTime);
+            if (!deadline) return;
+            startAfter = deadline > new Date() ? deadline : undefined;
+        } else {
+            const deadlineDay = new Date(`${task.endDate}T00:00:00Z`);
+            startAfter = deadlineDay > new Date() ? deadlineDay : undefined;
+        }
+
+        const data: DeadlineJobData = {
+            taskId: task.id,
+            description: task.description ?? '',
+            goalId: task.goalId,
+            goalListId: task.goalListId,
+            endDate: task.endDate,
+            endTime: task.endTime,
+            initiatorId: initiatorId ?? null,
+            immediate: !startAfter,
+        };
+
+        $logger.info(`[DeadlineScheduler] Scheduling task=${task.id} at=${startAfter?.toISOString() ?? 'immediate'}`);
+
+        await getJobQueue().send(DEADLINE_JOB, data, {
+            startAfter,
+            singletonKey: this.singletonKey(task.id),
+        });
+    }
+
+    async cancel(taskId: number): Promise<void> {
+        await cancelJobBySingletonKey(DEADLINE_JOB, this.singletonKey(taskId));
+    }
+
+    async registerWorker(): Promise<void> {
+        const boss = getJobQueue();
+        const db = Database.getInstance();
+
+        await boss.createQueue(DEADLINE_JOB);
+
+        await boss.work<DeadlineJobData>(DEADLINE_JOB, async ([job]) => {
+            const { taskId, description, goalId, goalListId, endDate, endTime, initiatorId, immediate } = job.data;
+
+            if (!taskId) return;
+            $logger.info(`[DeadlineScheduler] Worker: job=${job.id} task=${taskId}`);
+
+            const task = await db.dbDrizzle
+                .select({ owner: TasksSchema.owner, endDate: TasksSchema.endDate, endTime: TasksSchema.endTime })
+                .from(TasksSchema)
+                .where(and(eq(TasksSchema.id, taskId), or(eq(TasksSchema.complete, false), isNull(TasksSchema.complete))));
+
+            if (task.length === 0) {
+                $logger.info(`[DeadlineScheduler] Task ${taskId}: not found or completed`);
+                return;
+            }
+
+            if (task[0].endDate !== endDate || task[0].endTime !== endTime) {
+                $logger.info(`[DeadlineScheduler] Task ${taskId}: stale job, deadline changed`);
+                return;
+            }
+
+            const recipientIds = await this.resolveRecipients(db, taskId, goalId, task[0].owner);
+            if (!recipientIds || recipientIds.length === 0) {
+                $logger.info(`[DeadlineScheduler] Task ${taskId}: no recipients`);
+                return;
+            }
+
+            if (immediate && initiatorId) {
+                const idx = recipientIds.indexOf(initiatorId);
+                if (idx !== -1) recipientIds.splice(idx, 1);
+            }
+
+            if (recipientIds.length === 0) return;
+
+            $logger.info(`[DeadlineScheduler] Task ${taskId}: sending to [${recipientIds.join(',')}]`);
+
+            const tz = task[0].owner ? await this.deviceTokensRepo.getTimezoneByUserId(task[0].owner) : 'UTC';
+            const message = NotificationMessages.deadline(description, endDate, endTime, tz);
+
+            await getNotificationService().notifyMany(
+                recipientIds,
+                NotificationType.DEADLINE,
+                message,
+                { goalId, goalListId },
+                taskId,
+            );
+        });
+    }
+
+    private singletonKey(taskId: number): string {
+        return `deadline-${taskId}`;
+    }
+
+    private async resolveRecipients(db: Database, taskId: number, goalId: number, taskOwner: number | null): Promise<number[] | null> {
+        const authUsers = alias(UsersSchema, 'auth_users');
+
+        try {
+            const [assignees, goal] = await Promise.all([
+                db.dbDrizzle
+                    .select({ userId: authUsers.id })
+                    .from(TasksAssigneeSchema)
+                    .innerJoin(CollaborationUsersSchema, eq(TasksAssigneeSchema.collabUserId, CollaborationUsersSchema.id))
+                    .innerJoin(authUsers, eq(CollaborationUsersSchema.email, authUsers.email))
+                    .where(eq(TasksAssigneeSchema.taskId, taskId)),
+                db.dbDrizzle
+                    .select({ owner: GoalsSchema.owner })
+                    .from(GoalsSchema)
+                    .where(eq(GoalsSchema.id, goalId)),
+            ]);
+
+            const ids = new Set<number>();
+            if (taskOwner) ids.add(taskOwner);
+            assignees.forEach((r) => ids.add(r.userId));
+            if (goal[0]) ids.add(goal[0].owner);
+
+            return [...ids];
+        } catch (err) {
+            $logger.error(err, '[DeadlineScheduler] Failed to resolve recipients');
+            return null;
+        }
+    }
+}

api/src/tv-modules/notifications/types.ts

@@ -0,0 +1,123 @@
+export enum NotificationType {
+    DEADLINE = 'deadline',
+    ASSIGN = 'assign',
+    MENTION = 'mention',
+    COMMENT = 'comment',
+    STATUS_CHANGE = 'status_change',
+}
+
+export enum NotificationChannel {
+    PUSH = 'push',
+    WEBSOCKET = 'websocket',
+    EMAIL = 'email',
+}
+
+export interface NotificationMessage {
+    title: string;
+    body: string;
+}
+
+/** 
+ * Base settings — only channels (for instant notification types) 
+ */
+export interface BaseTypeSettings {
+    channels?: Partial<Record<NotificationChannel, boolean>>;
+}
+
+/** 
+ * Fixed intervals in minutes before deadline. 
+ * Key = minutes, value = enabled. undefined = enabled (opt-out) 
+ */
+export interface DeadlineIntervals {
+    0?: boolean;     // at deadline
+    15?: boolean;    // 15 min before
+    30?: boolean;    // 30 min before
+    60?: boolean;    // 1 hour before
+    1440?: boolean;  // 1 day before
+}
+
+/** 
+ * Deadline has intervals (minutes before deadline to notify) 
+ */
+export interface DeadlineTypeSettings extends BaseTypeSettings {
+    intervals?: DeadlineIntervals;
+}
+
+/** 
+ * Instant types — only channels, no intervals 
+ */
+export type InstantTypeSettings = BaseTypeSettings;
+
+/** 
+ * Maps each notification type to its allowed settings shape 
+ */
+export interface TypeSettingsMap {
+    [NotificationType.DEADLINE]: DeadlineTypeSettings;
+    [NotificationType.ASSIGN]: InstantTypeSettings;
+    [NotificationType.MENTION]: InstantTypeSettings;
+    [NotificationType.COMMENT]: InstantTypeSettings;
+    [NotificationType.STATUS_CHANGE]: InstantTypeSettings;
+}
+
+/** 
+ * Union of all possible type settings (for generic use) 
+ */
+export type TypeSettings = DeadlineTypeSettings | InstantTypeSettings;
+
+/**
+ * @example
+ * {
+ *   "global": {
+ *     "deadline": {
+ *       "channels": { "push": true, "websocket": true, "email": false },
+ *       "intervals": { "0": true, "15": true, "60": true, "1440": false }
+ *     },
+ *     "assign": {
+ *       "channels": { "push": true, "websocket": true }
+ *     },
+ *     "mention": {
+ *       "channels": { "push": false }
+ *     }
+ *   },
+ *   "projects": {
+ *     "42": {
+ *       "deadline": {
+ *         "channels": { "push": false },
+ *         "intervals": { "0": true, "30": true }
+ *       }
+ *     }
+ *   }
+ * }
+ *
+ * Result for user with these settings:
+ * - deadline globally: push + websocket, remind at 0/15/60 min before (1440 disabled)
+ * - deadline in project 42: websocket only (push overridden), remind at 0/30 min before
+ * - assign globally: push + websocket
+ * - mention globally: websocket only (push explicitly disabled)
+ * - comment: no settings → all channels enabled (opt-out)
+ */
+export interface SettingsJson {
+    global?: { [K in NotificationType]?: TypeSettingsMap[K] };
+    projects?: Record<string, { [K in NotificationType]?: TypeSettingsMap[K] }>;
+}
+
+export interface DeadlineJobData {
+    taskId: number;
+    description: string;
+    goalId: number;
+    goalListId: number | null;
+    endDate: string;
+    endTime: string | null;
+    initiatorId: number | null;
+    immediate: boolean;
+}
+
+export interface TaskWithDeadline {
+    id: number;
+    description: string | null;
+    goalId: number;
+    goalListId: number | null;
+    owner: number | null;
+    endDate: string | null;
+    endTime: string | null;
+}

api/src/tv-modules/notifications/utils.ts

@@ -0,0 +1,42 @@
+/**
+ * Parse UTC time string (HH:mm:ss) with date into a Date object
+ */
+export function parseUtcTime(dateStr: string, timeStr: string): Date | null {
+    const match = timeStr.match(/^(\d{2}):(\d{2})/);
+    if (!match) return null;
+    return new Date(`${dateStr}T${match[1]}:${match[2]}:00Z`);
+}
+
+/**
+ * Convert a local hour (e.g. 9 for 09:00) in a given IANA timezone
+ * to a UTC Date for the specified date string (YYYY-MM-DD)
+ */
+export function localHourToUtc(dateStr: string, hour: number, timezone: string): Date {
+    try {
+        const formatter = new Intl.DateTimeFormat('en-US', {
+            timeZone: timezone,
+            year: 'numeric', month: '2-digit', day: '2-digit',
+            hour: '2-digit', minute: '2-digit', second: '2-digit',
+            hour12: false,
+        });
+
+        const utcMidnight = new Date(`${dateStr}T00:00:00Z`);
+        const parts = formatter.formatToParts(utcMidnight);
+        const tzHour = Number(parts.find(p => p.type === 'hour')?.value ?? 0);
+        const tzDay = Number(parts.find(p => p.type === 'day')?.value ?? 0);
+        const utcDay = utcMidnight.getUTCDate();
+
+        let offsetHours = tzHour - utcMidnight.getUTCHours();
+        if (tzDay > utcDay) offsetHours += 24;
+        else if (tzDay < utcDay) offsetHours -= 24;
+
+        const result = new Date(`${dateStr}T00:00:00Z`);
+        result.setUTCHours(hour - offsetHours, 0, 0, 0);
+        return result;
+    } catch {
+        const fallback = new Date(`${dateStr}T00:00:00Z`);
+        fallback.setUTCHours(hour, 0, 0, 0);
+        return fallback;
+    }
+}
+

api/src/tv-modules/tasks/TasksController.ts

@@ -186,9 +186,9 @@ export class TasksController {
             return res.status(400).end();
         }
 
-        const subtasks = await req.appUser.tasksManager.fetchSubtasks(args.data);
+        // const subtasks = await req.appUser.tasksManager.fetchSubtasks(args.data);
 
-        return res.tvJson(subtasks);
+        return res.tvJson([]);
     };
 
     /** @deprecated */

api/src/tv-modules/tasks/TasksManager.ts

@@ -39,7 +39,7 @@ import {
     type TaskForClientNew,
     type TasksArgToggleTaskUsers,
 } from './tasks.server.types';
-import type { KanbanArgFetchTasksForColumn } from '../kanban/types';
+import type { KanbanArgFetchTasksForColumn, KanbanArgFilters } from '../kanban/types';
 
 type TaskFieldPermissionKey = keyof typeof TaskFieldPermissionsForEditOrCreation & keyof TasksSchemaTypeForSelect;
 
@@ -264,6 +264,13 @@ export class TasksManager {
     }
 
     async updateTask(data: TaskArgUpdate): Promise<{ task: TaskForClientNew; syncFailed?: boolean } | null> {
+        if (data.statusId !== undefined) {
+            const currentTask = await this.repository.fetchTaskByIdNew(data.id);
+            if (currentTask && currentTask.statusId !== data.statusId) {
+                data.kanbanOrder = await this.repository.getNextKanbanOrder(currentTask.goalId);
+            }
+        }
+
         const task = await this.repository.updateTask(data);
         if (!task) {
             return null;
@@ -279,7 +286,7 @@ export class TasksManager {
         eventBus.emit('task.updated', {
             task,
             changes,
-            userId: this.user.getUserData()?.id as number,
+            initiatorId: this.user.getUserData()?.id as number,
         });
 
         const tasks = await this.extendTasksWithTagsAndAssignees([task]);
@@ -401,7 +408,7 @@ export class TasksManager {
         if (task[0]) {
             eventBus.emit('task.created', {
                 task: task[0],
-                userId: this.user.getUserData()?.id as number,
+                initiatorId: this.user.getUserData()?.id as number,
             });
         }
 
@@ -409,7 +416,12 @@ export class TasksManager {
     }
 
     async deleteTaskNew(data: TaskArgDelete) {
-        return await this.repository.deleteTaskNew(data);
+        const task = await this.repository.fetchTaskByIdNew(data.taskId);
+        const result = await this.repository.deleteTaskNew(data);
+        if (result) {
+            eventBus.emit('task.deleted', { taskId: data.taskId, goalId: task?.goalId ?? 0, initiatorId: this.user.getUserData()?.id as number });
+        }
+        return result;
     }
 
     async toggleTaskUsers(data: TasksArgToggleTaskUsers) {
@@ -417,12 +429,13 @@ export class TasksManager {
         eventBus.emit('task.assigneesChanged', {
             taskId: data.taskId,
             userIds: data.userIds,
+            initiatorId: this.user.getUserData()?.id as number,
         });
         return result;
     }
 
-    async fetchTasksForKanbanColumn(data: KanbanArgFetchTasksForColumn): Promise<{ tasks: TaskForClientNew[], nextCursor: string | number | null }> {
-        const tasks = await this.repository.fetchTasksForKanbanColumn(data.goalId, data.columnId, data.cursor);
+    async fetchTasksForKanbanColumn(data: KanbanArgFetchTasksForColumn & { filters?: KanbanArgFilters }): Promise<{ tasks: TaskForClientNew[], nextCursor: string | number | null }> {
+        const tasks = await this.repository.fetchTasksForKanbanColumn(data.goalId, data.columnId, data.cursor, data.filters);
         if (!tasks || tasks.length === 0) return { tasks: [], nextCursor: null };
         return { tasks: await this.extendTasksWithTagsAndAssignees(tasks), nextCursor: tasks[tasks.length - 1].kanbanOrder };
     }

api/src/tv-modules/tasks/TasksRepository.ts

@@ -30,6 +30,7 @@ import type {
     TaskArgUpdate,
     TasksArgToggleTaskUsers,
 } from './tasks.server.types';
+import type { KanbanArgFilters } from '../kanban/types';
 
 export class TasksRepository {
     private readonly db: Database;
@@ -643,7 +644,7 @@ export class TasksRepository {
         return !!result?.rowCount;
     }
 
-    async fetchTasksForKanbanColumn(goalId: number, columnId: number | null, cursor: number | null): Promise<TasksSchemaTypeForSelect[]> {
+    async fetchTasksForKanbanColumn(goalId: number, columnId: number | null, cursor: number | null, filters?: KanbanArgFilters): Promise<TasksSchemaTypeForSelect[]> {
         const conditions = [
             eq(TasksSchema.goalId, goalId),
             columnId === null ? isNull(TasksSchema.statusId) : eq(TasksSchema.statusId, columnId),
@@ -653,6 +654,24 @@ export class TasksRepository {
         if (cursor !== null) {
             conditions.push(gt(TasksSchema.kanbanOrder, cursor));
         }
+        if (filters?.listIds && filters.listIds.length > 0) {
+            conditions.push(inArray(TasksSchema.goalListId, filters.listIds));
+        }
+        if (filters?.assigneeIds && filters.assigneeIds.length > 0) {
+            conditions.push(
+                exists(
+                    this.db.dbDrizzle
+                        .select({ one: sql`1` })
+                        .from(TasksAssigneeSchema)
+                        .where(
+                            and(
+                                eq(TasksAssigneeSchema.taskId, TasksSchema.id),
+                                inArray(TasksAssigneeSchema.collabUserId, filters.assigneeIds)
+                            )
+                        )
+                )
+            );
+        }
 
         const result = await callWithCatch(() =>
             this.db.dbDrizzle.select().from(TasksSchema).where(and(...conditions)).orderBy(asc(TasksSchema.kanbanOrder)).limit(20)
@@ -660,11 +679,13 @@ export class TasksRepository {
         return result ?? [];
     }
 
-    async fetchTaskWithMinKanbanOrder(goalId: number, columnId: number | null): Promise<number | null> {
+    async fetchTaskWithMinKanbanOrder(goalId: number, columnId?: number | null): Promise<number | null> {
         const conditions = [
             eq(TasksSchema.goalId, goalId),
-            columnId === null ? isNull(TasksSchema.statusId) : eq(TasksSchema.statusId, columnId),
         ];
+        if (columnId !== undefined) {
+            conditions.push(columnId === null ? isNull(TasksSchema.statusId) : eq(TasksSchema.statusId, columnId));
+        }
         const result = await callWithCatch(() => this.db.dbDrizzle.select({
             minKanbanOrder: sql<number>`MIN(kanban_order)`
         }).from(TasksSchema).where(and(...conditions)));
@@ -673,8 +694,8 @@ export class TasksRepository {
 
     static readonly KANBAN_ORDER_GAP = 16384;
 
-    async getNextKanbanOrder(goalId: number, statusId: number | null = null): Promise<number> {
-        const min = await this.fetchTaskWithMinKanbanOrder(goalId, statusId);
+    async getNextKanbanOrder(goalId: number, columnId?: number | null): Promise<number> {
+        const min = await this.fetchTaskWithMinKanbanOrder(goalId, columnId);
         return (min ?? 0) - TasksRepository.KANBAN_ORDER_GAP;
     }
 }

api/src/tv-modules/webhooks/WebhooksController.ts

@@ -0,0 +1,93 @@
+import type { Request, Response } from 'express';
+import { ArkErrors } from 'arktype';
+import { WebhooksManager } from './WebhooksManager';
+import {
+    WebhookArkTypeCreate,
+    WebhookArkTypeUpdate,
+    WebhookArkTypeDelete,
+    WebhookArkTypeFetch,
+    WebhookArkTypeById,
+    WebhookArkTypeFetchDeliveries,
+} from './types';
+
+export class WebhooksController {
+    private readonly manager = new WebhooksManager();
+
+    create = async (req: Request, res: Response) => {
+        const data = WebhookArkTypeCreate(req.body);
+        if (data instanceof ArkErrors) {
+            return res.status(400).send(data.summary);
+        }
+        const result = await this.manager.create(data);
+        if (!result) return res.status(500).end();
+        return res.tvJson(result);
+    };
+
+    update = async (req: Request, res: Response) => {
+        const data = WebhookArkTypeUpdate(req.body);
+        if (data instanceof ArkErrors) {
+            return res.status(400).send(data.summary);
+        }
+        const result = await this.manager.update(data);
+        if (!result) return res.status(404).end();
+        return res.tvJson(result);
+    };
+
+    delete = async (req: Request, res: Response) => {
+        const data = WebhookArkTypeDelete(req.body);
+        if (data instanceof ArkErrors) {
+            return res.status(400).send(data.summary);
+        }
+        const result = await this.manager.delete(data.id);
+        return res.tvJson(result);
+    };
+
+    fetch = async (req: Request, res: Response) => {
+        const data = WebhookArkTypeFetch(req.query);
+        if (data instanceof ArkErrors) {
+            return res.status(400).send(data.summary);
+        }
+        const result = await this.manager.fetchByGoalId(data.goalId);
+        return res.tvJson(result);
+    };
+
+    rotateSecret = async (req: Request, res: Response) => {
+        const data = WebhookArkTypeById(req.body);
+        if (data instanceof ArkErrors) {
+            return res.status(400).send(data.summary);
+        }
+        const result = await this.manager.rotateSecret(data.id);
+        if (!result) return res.status(404).end();
+        return res.tvJson(result);
+    };
+
+    testDelivery = async (req: Request, res: Response) => {
+        const data = WebhookArkTypeById(req.body);
+        if (data instanceof ArkErrors) {
+            return res.status(400).send(data.summary);
+        }
+        const result = await this.manager.testDelivery(data.id);
+        return res.tvJson(result);
+    };
+
+    fetchDeliveries = async (req: Request, res: Response) => {
+        const data = WebhookArkTypeFetchDeliveries({ ...req.params, ...req.query });
+        if (data instanceof ArkErrors) {
+            return res.status(400).send(data.summary);
+        }
+        const result = await this.manager.fetchDeliveries(data.id, {
+            cursor: data.cursor,
+            status: data.status,
+        });
+        return res.tvJson(result);
+    };
+
+    retryDelivery = async (req: Request, res: Response) => {
+        const data = WebhookArkTypeById(req.body);
+        if (data instanceof ArkErrors) {
+            return res.status(400).send(data.summary);
+        }
+        const result = await this.manager.retryDelivery(data.id);
+        return res.tvJson(result);
+    };
+}

api/src/tv-modules/webhooks/WebhooksDispatcher.ts

@@ -0,0 +1,102 @@
+import { eq } from 'drizzle-orm';
+import { TasksSchema } from 'taskview-db-schemas';
+import { eventBus, type AppEvents } from '../../core/EventBus';
+import { getJobQueue } from '../../core/JobQueue';
+import { decrypt } from '../../utils/crypto';
+import { $logger } from '../../modules/logget';
+import { Database } from '../../modules/db';
+import { WebhooksRepository } from './WebhooksRepository';
+import { WebhooksManager } from './WebhooksManager';
+import type { WebhookDeliverJobData } from './types';
+import type { Dispatcher } from '../../core/Dispatcher';
+
+const WEBHOOK_DELIVER_JOB = 'webhook-deliver';
+const MAX_ATTEMPTS = 3;
+const MAX_CONSECUTIVE_FAILURES = 10;
+
+export class WebhooksDispatcher implements Dispatcher {
+    private readonly repository = new WebhooksRepository();
+    private readonly manager = new WebhooksManager();
+
+    register(): void {
+        eventBus.on('task.created', (data) => this.dispatch('task.created', data.task.goalId, data));
+        eventBus.on('task.updated', (data) => this.dispatch('task.updated', data.task.goalId, data));
+        eventBus.on('task.deleted', (data) => this.dispatch('task.deleted', data.goalId, data));
+        eventBus.on('task.assigneesChanged', (data) => this.dispatchAssigneesChanged(data));
+    }
+
+    async registerWorkers(): Promise<void> {
+        const boss = getJobQueue();
+        await boss.createQueue(WEBHOOK_DELIVER_JOB);
+        await boss.work<WebhookDeliverJobData>(WEBHOOK_DELIVER_JOB, async ([job]) => {
+            await this.deliverJob(job.data);
+        });
+    }
+
+    private async dispatch(event: string, goalId: number, payload: object): Promise<void> {
+        const webhooks = await this.repository.fetchActiveByGoalIdAndEvent(goalId, event);
+        for (const webhook of webhooks) {
+            await this.enqueueDelivery(webhook.id, webhook.url, webhook.secretEncrypted, event, {
+                event,
+                timestamp: new Date().toISOString(),
+                ...payload,
+            });
+        }
+    }
+
+    private async dispatchAssigneesChanged(data: AppEvents['task.assigneesChanged']): Promise<void> {
+        const db = Database.getInstance();
+        const task = await db.dbDrizzle.select().from(TasksSchema).where(eq(TasksSchema.id, data.taskId)).limit(1);
+        if (!task[0]) return;
+        await this.dispatch('task.assigneesChanged', task[0].goalId, data);
+    }
+
+    private async enqueueDelivery(webhookId: number, url: string, secretEncrypted: string, event: string, payload: object): Promise<void> {
+        const delivery = await this.repository.createDelivery({ webhookId, event, payload });
+        if (!delivery) return;
+
+        const boss = getJobQueue();
+        await boss.send(WEBHOOK_DELIVER_JOB, {
+            deliveryId: delivery.id,
+            webhookId,
+            url,
+            secretEncrypted,
+            payload,
+            attempt: 1,
+        } satisfies WebhookDeliverJobData);
+    }
+
+    private async deliverJob(data: WebhookDeliverJobData): Promise<void> {
+        const secret = decrypt(data.secretEncrypted);
+        const result = await this.manager.deliver(data.url, secret, data.payload);
+
+        await this.repository.updateDelivery(data.deliveryId, {
+            status: result.success ? 'success' : (data.attempt >= MAX_ATTEMPTS ? 'failed' : 'pending'),
+            responseCode: result.responseCode,
+            attempts: data.attempt,
+        });
+
+        if (result.success) {
+            await this.repository.resetConsecutiveFailures(data.webhookId);
+            return;
+        }
+
+        if (data.attempt < MAX_ATTEMPTS) {
+            const boss = getJobQueue();
+            const delay = Math.pow(2, data.attempt) * 5;
+            await boss.send(WEBHOOK_DELIVER_JOB, {
+                ...data,
+                attempt: data.attempt + 1,
+            }, { startAfter: delay });
+            return;
+        }
+
+        const failures = await this.repository.incrementConsecutiveFailures(data.webhookId);
+        if (failures >= MAX_CONSECUTIVE_FAILURES) {
+            await this.repository.deactivate(data.webhookId);
+            $logger.warn(`[Webhooks] Deactivated webhook=${data.webhookId} after ${failures} consecutive failures`);
+        } else {
+            $logger.warn(`[Webhooks] Delivery failed for webhook=${data.webhookId}, consecutive failures: ${failures}/${MAX_CONSECUTIVE_FAILURES}`);
+        }
+    }
+}

api/src/tv-modules/webhooks/WebhooksManager.ts

@@ -0,0 +1,124 @@
+import { randomBytes, createHmac } from 'crypto';
+import { encrypt, decrypt } from '../../utils/crypto';
+import { $logger } from '../../modules/logget';
+import { WebhooksRepository } from './WebhooksRepository';
+import type { WebhookArgCreate, WebhookArgUpdate } from './types';
+import type { WebhooksSchemaTypeForSelect } from 'taskview-db-schemas';
+
+export type WebhookForClient = Omit<WebhooksSchemaTypeForSelect, 'secretEncrypted'>;
+
+export class WebhooksManager {
+    public readonly repository: WebhooksRepository;
+
+    constructor() {
+        this.repository = new WebhooksRepository();
+    }
+
+    async create(data: WebhookArgCreate): Promise<{ webhook: WebhookForClient; secret: string } | null> {
+        const secret = randomBytes(32).toString('hex');
+        const secretEncrypted = encrypt(secret);
+
+        const webhook = await this.repository.create({
+            goalId: data.goalId,
+            url: data.url,
+            secretEncrypted,
+            events: data.events,
+        });
+
+        if (!webhook) return null;
+
+        return { webhook: this.toClient(webhook), secret };
+    }
+
+    async update(data: WebhookArgUpdate): Promise<WebhookForClient | null> {
+        const updateData: Partial<{ url: string; events: string[]; isActive: boolean }> = {};
+        if (data.url !== undefined) updateData.url = data.url;
+        if (data.events !== undefined) updateData.events = data.events;
+        if (data.isActive !== undefined) updateData.isActive = data.isActive;
+
+        const webhook = await this.repository.update(data.id, updateData);
+        if (!webhook) return null;
+        return this.toClient(webhook);
+    }
+
+    async delete(id: number): Promise<boolean> {
+        return this.repository.delete(id);
+    }
+
+    async fetchByGoalId(goalId: number): Promise<WebhookForClient[]> {
+        const webhooks = await this.repository.fetchByGoalId(goalId);
+        return webhooks.map(w => this.toClient(w));
+    }
+
+    async rotateSecret(id: number): Promise<{ secret: string } | null> {
+        const secret = randomBytes(32).toString('hex');
+        const secretEncrypted = encrypt(secret);
+        const success = await this.repository.updateSecret(id, secretEncrypted);
+        if (!success) return null;
+        return { secret };
+    }
+
+    async testDelivery(id: number): Promise<{ success: boolean; responseCode?: number }> {
+        const webhook = await this.repository.fetchById(id);
+        if (!webhook) return { success: false };
+
+        const secret = decrypt(webhook.secretEncrypted);
+        const payload = {
+            event: 'webhook.test',
+            timestamp: new Date().toISOString(),
+            data: { message: 'This is a test webhook delivery' },
+        };
+
+        return this.deliver(webhook.url, secret, payload);
+    }
+
+    async deliver(url: string, secret: string, payload: object): Promise<{ success: boolean; responseCode?: number }> {
+        const body = JSON.stringify(payload);
+        const signature = createHmac('sha256', secret).update(body).digest('hex');
+
+        try {
+            const response = await fetch(url, {
+                method: 'POST',
+                headers: {
+                    'Content-Type': 'application/json',
+                    'X-Webhook-Signature': `sha256=${signature}`,
+                },
+                body,
+                signal: AbortSignal.timeout(10000),
+            });
+
+            return { success: response.ok, responseCode: response.status };
+        } catch (err) {
+            $logger.error(err, `[Webhooks] Delivery failed to ${url}`);
+            return { success: false };
+        }
+    }
+
+    async retryDelivery(deliveryId: number): Promise<{ success: boolean; responseCode?: number }> {
+        const deliveries = await this.repository.fetchDeliveryById(deliveryId);
+        if (!deliveries) return { success: false };
+
+        const webhook = await this.repository.fetchById(deliveries.webhookId);
+        if (!webhook) return { success: false };
+
+        const secret = decrypt(webhook.secretEncrypted);
+        const result = await this.deliver(webhook.url, secret, deliveries.payload as object);
+
+        await this.repository.updateDelivery(deliveryId, {
+            status: result.success ? 'success' : 'failed',
+            responseCode: result.responseCode,
+            attempts: deliveries.attempts + 1,
+        });
+
+        return result;
+    }
+
+    async fetchDeliveries(webhookId: number, options?: { cursor?: number; status?: string }) {
+        return this.repository.fetchDeliveries(webhookId, options);
+    }
+
+    private toClient(webhook: WebhooksSchemaTypeForSelect): WebhookForClient {
+        const { secretEncrypted, ...rest } = webhook;
+        return rest;
+    }
+}

api/src/tv-modules/webhooks/WebhooksRepository.ts

@@ -0,0 +1,144 @@
+import { and, desc, eq, lt, sql } from 'drizzle-orm';
+import { WebhooksSchema, WebhookDeliveriesSchema, type WebhooksSchemaTypeForSelect, type WebhookDeliveriesSchemaTypeForSelect } from 'taskview-db-schemas';
+import { Database } from '../../modules/db';
+import { callWithCatch } from '../../utils/helpers';
+
+export class WebhooksRepository {
+    private readonly db: Database;
+
+    constructor() {
+        this.db = Database.getInstance();
+    }
+
+    async create(data: { goalId: number; url: string; secretEncrypted: string; events: string[] }): Promise<WebhooksSchemaTypeForSelect | null> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.insert(WebhooksSchema).values(data).returning()
+        );
+        return result?.[0] ?? null;
+    }
+
+    async update(id: number, data: Partial<{ url: string; events: string[]; isActive: boolean }>): Promise<WebhooksSchemaTypeForSelect | null> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.update(WebhooksSchema)
+                .set({ ...data, updatedAt: new Date() })
+                .where(eq(WebhooksSchema.id, id))
+                .returning()
+        );
+        return result?.[0] ?? null;
+    }
+
+    async updateSecret(id: number, secretEncrypted: string): Promise<boolean> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.update(WebhooksSchema)
+                .set({ secretEncrypted, updatedAt: new Date() })
+                .where(eq(WebhooksSchema.id, id))
+                .returning()
+        );
+        return (result?.length ?? 0) > 0;
+    }
+
+    async delete(id: number): Promise<boolean> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.delete(WebhooksSchema).where(eq(WebhooksSchema.id, id))
+        );
+        return !!result?.rowCount;
+    }
+
+    async fetchById(id: number): Promise<WebhooksSchemaTypeForSelect | null> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.select().from(WebhooksSchema).where(eq(WebhooksSchema.id, id))
+        );
+        return result?.[0] ?? null;
+    }
+
+    async fetchByGoalId(goalId: number): Promise<WebhooksSchemaTypeForSelect[]> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.select().from(WebhooksSchema).where(eq(WebhooksSchema.goalId, goalId))
+        );
+        return result ?? [];
+    }
+
+    async fetchActiveByGoalIdAndEvent(goalId: number, event: string): Promise<WebhooksSchemaTypeForSelect[]> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.select().from(WebhooksSchema).where(
+                and(
+                    eq(WebhooksSchema.goalId, goalId),
+                    eq(WebhooksSchema.isActive, true),
+                )
+            )
+        );
+        return (result ?? []).filter(w => w.events.includes(event));
+    }
+
+    async createDelivery(data: { webhookId: number; event: string; payload: unknown }): Promise<WebhookDeliveriesSchemaTypeForSelect | null> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.insert(WebhookDeliveriesSchema).values({
+                webhookId: data.webhookId,
+                event: data.event,
+                payload: data.payload,
+            }).returning()
+        );
+        return result?.[0] ?? null;
+    }
+
+    async updateDelivery(id: number, data: { status: string; responseCode?: number; attempts: number }): Promise<void> {
+        await callWithCatch(() =>
+            this.db.dbDrizzle.update(WebhookDeliveriesSchema)
+                .set({ ...data, lastAttemptAt: new Date() })
+                .where(eq(WebhookDeliveriesSchema.id, id))
+        );
+    }
+
+    async fetchDeliveryById(id: number): Promise<WebhookDeliveriesSchemaTypeForSelect | null> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.select().from(WebhookDeliveriesSchema).where(eq(WebhookDeliveriesSchema.id, id))
+        );
+        return result?.[0] ?? null;
+    }
+
+    async fetchDeliveries(webhookId: number, options?: { cursor?: number; status?: string; limit?: number }): Promise<WebhookDeliveriesSchemaTypeForSelect[]> {
+        const limit = options?.limit ?? 20;
+        const conditions = [eq(WebhookDeliveriesSchema.webhookId, webhookId)];
+
+        if (options?.cursor) {
+            conditions.push(lt(WebhookDeliveriesSchema.id, options.cursor));
+        }
+        if (options?.status) {
+            conditions.push(eq(WebhookDeliveriesSchema.status, options.status));
+        }
+
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.select().from(WebhookDeliveriesSchema)
+                .where(and(...conditions))
+                .orderBy(desc(WebhookDeliveriesSchema.id))
+                .limit(limit)
+        );
+        return result ?? [];
+    }
+
+    async resetConsecutiveFailures(webhookId: number): Promise<void> {
+        await callWithCatch(() =>
+            this.db.dbDrizzle.update(WebhooksSchema)
+                .set({ consecutiveFailures: 0 })
+                .where(eq(WebhooksSchema.id, webhookId))
+        );
+    }
+
+    async incrementConsecutiveFailures(webhookId: number): Promise<number> {
+        const result = await callWithCatch(() =>
+            this.db.dbDrizzle.update(WebhooksSchema)
+                .set({ consecutiveFailures: sql`${WebhooksSchema.consecutiveFailures} + 1` })
+                .where(eq(WebhooksSchema.id, webhookId))
+                .returning({ consecutiveFailures: WebhooksSchema.consecutiveFailures })
+        );
+        return result?.[0]?.consecutiveFailures ?? 0;
+    }
+
+    async deactivate(webhookId: number): Promise<void> {
+        await callWithCatch(() =>
+            this.db.dbDrizzle.update(WebhooksSchema)
+                .set({ isActive: false, updatedAt: new Date() })
+                .where(eq(WebhooksSchema.id, webhookId))
+        );
+    }
+}

api/src/tv-modules/webhooks/WebhooksRoutes.ts

@@ -0,0 +1,30 @@
+import { Router } from 'express';
+import type { Routable } from '../../types/routable.type';
+import { IsLoggedIn } from '../auth/middlewares/is-logged-in';
+import { WebhooksController } from './WebhooksController';
+
+export default class WebhooksRoutes implements Routable {
+    private readonly router: ReturnType<typeof Router>;
+    private readonly controller: WebhooksController;
+
+    constructor() {
+        this.router = Router();
+        this.controller = new WebhooksController();
+        this.initRoutes();
+    }
+
+    getRouter() {
+        return this.router;
+    }
+
+    initRoutes() {
+        this.router.get('', [IsLoggedIn], this.controller.fetch);
+        this.router.post('', [IsLoggedIn], this.controller.create);
+        this.router.patch('', [IsLoggedIn], this.controller.update);
+        this.router.delete('', [IsLoggedIn], this.controller.delete);
+        this.router.post('/rotate-secret', [IsLoggedIn], this.controller.rotateSecret);
+        this.router.post('/test', [IsLoggedIn], this.controller.testDelivery);
+        this.router.get('/deliveries/:id', [IsLoggedIn], this.controller.fetchDeliveries);
+        this.router.post('/retry', [IsLoggedIn], this.controller.retryDelivery);
+    }
+}

api/src/tv-modules/webhooks/types.ts

@@ -0,0 +1,66 @@
+import { type } from 'arktype';
+
+const NumberFromString = type('string|number').pipe((v) => Number(v));
+
+export const WebhookArkTypeCreate = type({
+    goalId: 'number',
+    url: 'string',
+    events: 'string[]',
+});
+
+export type WebhookArgCreate = typeof WebhookArkTypeCreate.infer;
+
+export const WebhookArkTypeUpdate = type({
+    id: 'number',
+    'url?': 'string',
+    'events?': 'string[]',
+    'isActive?': 'boolean',
+});
+
+export type WebhookArgUpdate = typeof WebhookArkTypeUpdate.infer;
+
+export const WebhookArkTypeDelete = type({
+    id: 'number',
+});
+
+export type WebhookArgDelete = typeof WebhookArkTypeDelete.infer;
+
+export const WebhookArkTypeFetch = type({
+    goalId: NumberFromString,
+});
+
+export type WebhookArgFetch = typeof WebhookArkTypeFetch.infer;
+
+export const WebhookArkTypeById = type({
+    id: NumberFromString,
+});
+
+export type WebhookArgById = typeof WebhookArkTypeById.infer;
+
+const OptionalNumberFromString = type('string|number|undefined').pipe((v) => v === undefined ? undefined : Number(v));
+
+export const WebhookArkTypeFetchDeliveries = type({
+    id: NumberFromString,
+    'cursor?': OptionalNumberFromString,
+    'status?': 'string',
+});
+
+export type WebhookArgFetchDeliveries = typeof WebhookArkTypeFetchDeliveries.infer;
+
+export const WEBHOOK_EVENTS = [
+    'task.created',
+    'task.updated',
+    'task.deleted',
+    'task.assigneesChanged',
+] as const;
+
+export type WebhookEvent = typeof WEBHOOK_EVENTS[number];
+
+export interface WebhookDeliverJobData {
+    deliveryId: number;
+    webhookId: number;
+    url: string;
+    secretEncrypted: string;
+    payload: object;
+    attempt: number;
+}