#Tasks API

import { authPlugin, type KeyPairSettings } from '@dotpress/auth'
import { createCache, type CacheStore } from '@dotpress/cache'
import { cachePlugin } from '@dotpress/cache/plugin'
import { createApp, type HttpAdapter, type Logger } from '@dotpress/core'
import type { EventBus } from '@dotpress/events'
import { eventsPlugin } from '@dotpress/events/plugin'
import { openapiPlugin } from '@dotpress/openapi'
import { createAuthService } from '../features/auth/auth.ts'
import type { Repositories } from '../services/index.ts'
import { routers } from './routes.ts'
import { servicesPlugin } from './services-plugin.ts'

export interface TasksAppOptions<TAdapter extends HttpAdapter> {
  adapter: TAdapter
  /** The repositories of `createRepositories()`, over one database. */
  repositories: Repositories
  keys: KeyPairSettings
  /** Public URL of the API: the issuer of the tokens. */
  publicUrl: string
  cacheStore: CacheStore
  /** The event bus of `services/bus.ts`, or a fake bus in tests. */
  events: EventBus
  /** bcrypt cost: lower in the tests. Defaults to 12. */
  passwordHashRounds?: number
  logger?: Logger
}

/**
 * The HTTP app, on any adapter: Fastify in `server.ts`, the memory adapter
 * of `@dotpress/testing` in the tests. Its services come from the caller:
 * the entry point in production, each test in the tests.
 */
export const createTasksApp = <TAdapter extends HttpAdapter>(
  options: TasksAppOptions<TAdapter>
) => {
  const { adapter, repositories, logger } = options
  const auth = createAuthService({
    db: repositories.db,
    users: repositories.users,
    keys: options.keys,
    issuer: options.publicUrl,
    passwordHashRounds: options.passwordHashRounds,
  })
  const cache = createCache({
    store: options.cacheStore,
    prefix: 'tasks-api',
    // A slow store counts as a failure: the request goes on without cache.
    storeTimeout: '200ms',
  })

  return createApp({
    adapter,
    ...(logger && { logger }),
    routers,
    plugins: [
      // ctx.services: the repositories, the memos and auth of the app.
      servicesPlugin({ ...repositories, auth }),
      // ctx.user and ctx.auth; /auth/login, /refresh, /me, /logout.
      authPlugin(auth),
      // ctx.memo (per request) and ctx.cache.
      cachePlugin({ cache }),
      // ctx.events: emits with the request logger and request ID.
      eventsPlugin(options.events),
      // /openapi.json and /docs: the routes guarded by requireAuth are
      // marked as secured.
      openapiPlugin({
        info: {
          title: 'Tasks API',
          version: '1.0.0',
          description:
            'A team task manager built with Dotpress. Sign up with `POST /users`, log in with `POST /auth/login`, then use the access token with the Authorize button.',
        },
        tags: [
          { name: 'auth', description: 'Sessions' },
          { name: 'users', description: 'Accounts' },
          { name: 'projects', description: 'Projects, members and dashboards' },
          { name: 'tasks', description: 'Tasks of the projects' },
          { name: 'health', description: 'Monitoring' },
        ],
      }),
    ],
  })
}

import { declareContextKey, definePlugin } from '@dotpress/core'
import type { AppServices } from '../services/index.ts'

// ctx.services, typed in every handler and middleware: the IDE completes
// its repositories and their methods. Module augmentation applies to the
// whole program, like the plugins of Dotpress do for ctx.mailer or
// ctx.cache.
declare module '@dotpress/core' {
  interface RequestContext {
    /** The services of the app (src/services): repositories, memos, auth. */
    services: AppServices
  }
}

// Using ctx.services in an app without the plugin throws an explicit error.
declareContextKey('services', 'servicesPlugin')

/**
 * Exposes the services of the app to the routes: the route files are plain
 * modules, without singletons, and the tests give their own services.
 */
export const servicesPlugin = (services: AppServices) =>
  definePlugin({
    name: 'services',
    setup(api) {
      api.decorateContext('services', () => services)
    },
  })

import { defineMemo } from '@dotpress/cache'
import {
  defineMiddleware,
  notFoundError,
  unauthorizedError,
  type RequestContext,
} from '@dotpress/core'
import type { ProjectRepository } from './repository.ts'

/**
 * The role of a user in a project, loaded once per request (`ctx.memo`):
 * the middleware, the handlers and the checks of the assignees share it.
 */
export const roleMemo = (projects: ProjectRepository) =>
  defineMemo({
    name: 'project-role',
    load: (projectId: string, userId: string) =>
      projects.findRole(projectId, userId),
  })

/**
 * The role of the current user in a project, or undefined for a
 * non-member: the routes then answer 404, so that the existence of the
 * project is not revealed.
 */
export const roleIn = (
  ctx: Pick<RequestContext, 'memo' | 'services'> & { user: { id: string } },
  projectId: string
) => ctx.memo(ctx.services.loaders.role, projectId, ctx.user.id)

/**
 * Middleware of the routes of a project (`/projects/:projectId/...`): a
 * non-member gets 404. The role is memoized, so the handlers read it again
 * without a query.
 */
export const requireMember = defineMiddleware(async (ctx) => {
  const { projectId } = ctx.req.params
  const { user } = ctx
  if (!user) return unauthorizedError()
  if (!projectId || !(await roleIn({ ...ctx, user }, projectId))) {
    return notFoundError('Project not found')
  }
})

import { requireAuth } from '@dotpress/auth'
import {
  createRouter,
  notFoundError,
  unprocessableError,
  type RequestContext,
} from '@dotpress/core'
import { z } from 'zod'
import { requireMember, roleIn } from '../projects/access.ts'
import { projectTag } from '../projects/stats.ts'
import { personOf, PersonView } from '../users/loaders.ts'
import { taskAssigned, taskCompleted } from './events.ts'
import type { Task } from './repository.ts'

const Status = z.enum(['todo', 'doing', 'done'])

const TaskView = z
  .object({
    id: z.uuid(),
    projectId: z.uuid(),
    title: z.string(),
    status: Status,
    assignee: PersonView.nullable(),
    createdAt: z.iso.datetime(),
    updatedAt: z.iso.datetime(),
  })
  .meta({ id: 'Task' })

const Title = z.string().trim().min(1).max(200)

const TASK_NOT_FOUND = 'Task not found'
const NOT_A_MEMBER = 'The assignee is not a member of the project'

/** What the helpers read in the context of a request. */
type Context = Pick<RequestContext, 'memo' | 'services'>

/** A task as the API returns it, with the name of its assignee. */
const toView = async ({ memo, services }: Context, task: Task) => ({
  id: task.id,
  projectId: task.projectId,
  title: task.title,
  status: task.status,
  assignee: await personOf(memo, services.loaders.user, task.assigneeId),
  createdAt: task.createdAt,
  updatedAt: task.updatedAt,
})

/** An assignee must be a member of the project (memoized check). */
const isMember = async (
  { memo, services }: Context,
  projectId: string,
  userId: string
) => (await memo(services.loaders.role, projectId, userId)) !== undefined

/**
 * The routes of the tasks: a plain module. The handlers find the services
 * of the app in ctx.services.
 */
export const tasksRouter = createRouter()

// /projects/:projectId/tasks: the tasks of a project, for its members.
const projectTasks = tasksRouter.createGroup('/projects/:projectId/tasks', [
  requireAuth,
  requireMember,
])

projectTasks.defineRoute({
  method: 'post',
  path: '/',
  status: 201,
  schema: (z) => ({
    body: z.object({ title: Title, assigneeId: z.uuid().optional() }),
    response: TaskView,
  }),
  meta: {
    summary: 'Create a task',
    tags: ['tasks'],
    responses: { 404: 'Project not found', 422: NOT_A_MEMBER },
  },
  // ctx.events: eventsPlugin, with the request ID in the metadata.
  handler: async (ctx) => {
    const { req, user, cache, events, services } = ctx
    const { projectId } = req.params
    const { title, assigneeId } = req.body
    if (assigneeId && !(await isMember(ctx, projectId, assigneeId))) {
      return unprocessableError(NOT_A_MEMBER)
    }

    const task = await services.tasks.create({
      projectId,
      title,
      assigneeId: assigneeId ?? null,
      createdBy: user.id,
    })
    await cache.invalidateTags([projectTag(projectId)])
    // After the write: the event describes something that happened. The
    // response does not wait for the email, sent by a listener.
    if (assigneeId && assigneeId !== user.id) {
      await events.emit(taskAssigned, {
        taskId: task.id,
        assigneeId,
        assignedBy: user.id,
      })
    }
    return toView(ctx, task)
  },
})

projectTasks.defineRoute({
  path: '/',
  schema: (z) => ({
    query: z.object({
      status: Status.optional(),
      assignee: z.uuid().optional(),
    }),
    response: z.array(TaskView),
  }),
  meta: {
    summary: 'List the tasks of a project',
    tags: ['tasks'],
    responses: { 404: 'Project not found' },
  },
  handler: async (ctx) => {
    const { req, services } = ctx
    const list = await services.tasks.listByProject(req.params.projectId, {
      status: req.query.status,
      assigneeId: req.query.assignee,
    })
    // The assignees of every task in one query: the memo batches them.
    return Promise.all(list.map((task) => toView(ctx, task)))
  },
})

// /tasks/:id: the project is not in the path, so the membership is checked
// in the handlers, once the task is loaded.
const taskGroup = tasksRouter.createGroup('/tasks', [requireAuth])

taskGroup.defineRoute({
  path: '/:id',
  schema: () => ({ response: TaskView }),
  meta: {
    summary: 'Get a task',
    tags: ['tasks'],
    responses: { 404: TASK_NOT_FOUND },
  },
  handler: async (ctx) => {
    const task = await ctx.services.tasks.findById(ctx.req.params.id)
    if (!task || !(await roleIn(ctx, task.projectId))) {
      return notFoundError(TASK_NOT_FOUND)
    }
    return toView(ctx, task)
  },
})

taskGroup.defineRoute({
  method: 'patch',
  path: '/:id',
  schema: (z) => ({
    body: z.object({
      title: Title.optional(),
      status: Status.optional(),
      // null: unassigned.
      assigneeId: z.uuid().nullable().optional(),
    }),
    response: TaskView,
  }),
  meta: {
    summary: 'Update a task: title, status, assignee',
    tags: ['tasks'],
    responses: { 404: TASK_NOT_FOUND, 422: NOT_A_MEMBER },
  },
  handler: async (ctx) => {
    const { req, user, cache, events, services } = ctx
    const task = await services.tasks.findById(req.params.id)
    if (!task || !(await roleIn(ctx, task.projectId))) {
      return notFoundError(TASK_NOT_FOUND)
    }
    const { assigneeId } = req.body
    if (assigneeId && !(await isMember(ctx, task.projectId, assigneeId))) {
      return unprocessableError(NOT_A_MEMBER)
    }

    const updated = (await services.tasks.update(task.id, req.body))!
    await cache.invalidateTags([projectTag(task.projectId)])

    if (
      updated.assigneeId &&
      updated.assigneeId !== task.assigneeId &&
      updated.assigneeId !== user.id
    ) {
      await events.emit(taskAssigned, {
        taskId: task.id,
        assigneeId: updated.assigneeId,
        assignedBy: user.id,
      })
    }
    if (updated.status === 'done' && task.status !== 'done') {
      await events.emit(taskCompleted, {
        taskId: task.id,
        completedBy: user.id,
      })
    }
    return toView(ctx, updated)
  },
})

import { defineListener } from '@dotpress/events'
import type { Mailer } from '@dotpress/mailer'
import { get, run } from '../../db/database.ts'
import type { Repositories } from '../../services/index.ts'
import { taskAssigned, taskCompleted } from '../tasks/events.ts'
import { taskAssignedMail, taskCompletedMail } from './mails.ts'

/**
 * The listeners of the task events: they email the people concerned. They
 * run in the worker (with Redis), or in the API process (without): both
 * create them with their own services.
 */
export const createNotifications = ({
  repositories: { db, users, projects, tasks },
  mailer,
}: {
  repositories: Repositories
  mailer: Mailer
}) => {
  /**
   * Sends once per message: queues deliver a message at least once, and a
   * retry after a timeout must not send a second email. A crash between
   * the email and the insert can still send one twice, never lose one.
   */
  const once = async (messageId: string, send: () => Promise<unknown>) => {
    if (get(db, 'SELECT 1 FROM handled_messages WHERE id = ?', messageId)) {
      return
    }
    await send()
    run(
      db,
      'INSERT INTO handled_messages (id, handled_at) VALUES (?, ?)',
      messageId,
      new Date().toISOString()
    )
  }

  const notifyAssignee = defineListener(taskAssigned, {
    name: 'notify-assignee',
    handle: async ({ payload, id, logger }) => {
      const [task, assignee, actor] = await Promise.all([
        tasks.findById(payload.taskId),
        users.findById(payload.assigneeId),
        users.findById(payload.assignedBy),
      ])
      // Reassigned or deleted since the event: nothing to say any more.
      if (!task || !assignee || !actor || task.assigneeId !== assignee.id) {
        return
      }
      const project = await projects.findById(task.projectId)
      await once(id, () =>
        mailer.withLogger(logger).send(taskAssignedMail, {
          to: { name: assignee.name, address: assignee.email },
          locale: assignee.locale,
          data: {
            recipientName: assignee.name,
            actorName: actor.name,
            taskTitle: task.title,
            projectName: project?.name ?? '',
          },
        })
      )
    },
  })

  const notifyCreator = defineListener(taskCompleted, {
    name: 'notify-creator',
    handle: async ({ payload, id, logger }) => {
      const task = await tasks.findById(payload.taskId)
      // Reopened since the event, or completed by its creator.
      if (
        !task ||
        task.status !== 'done' ||
        task.createdBy === payload.completedBy
      ) {
        return
      }
      const [creator, actor, project] = await Promise.all([
        users.findById(task.createdBy),
        users.findById(payload.completedBy),
        projects.findById(task.projectId),
      ])
      if (!creator || !actor) return
      await once(id, () =>
        mailer.withLogger(logger).send(taskCompletedMail, {
          to: { name: creator.name, address: creator.email },
          locale: creator.locale,
          data: {
            recipientName: creator.name,
            actorName: actor.name,
            taskTitle: task.title,
            projectName: project?.name ?? '',
          },
        })
      )
    },
  })

  return [notifyAssignee, notifyCreator]
}

import { createConsoleLogger } from '@dotpress/core'
import { createWorker } from '@dotpress/events'
import { createBullmqSource } from '@dotpress/events/bullmq'
import { loadConfig, requireInProduction } from './config.ts'
import { openDatabase } from './db/database.ts'
import { createNotifications } from './features/notifications/listeners.ts'
import { redisConnection } from './services/bus.ts'
import { createRepositories } from './services/index.ts'
import { createAppMailer, createMailProvider } from './services/mailer.ts'

// Entry of `dotpress-events work src/worker.ts` (npm run worker): reads the
// BullMQ queue and runs the listeners, until Ctrl+C or SIGTERM, which wait
// for the jobs in progress. It opens the same SQLite file as the API.
const config = loadConfig()
requireInProduction(config, ['SMTP_URL'])
if (!config.REDIS_URL) {
  throw new Error(
    'REDIS_URL is required: without it, the API runs the listeners itself.'
  )
}

export const logger = createConsoleLogger({ bindings: { process: 'worker' } })

// The services of the worker: the database and the mailer only.
const repositories = createRepositories(openDatabase(config.DATABASE_PATH))
const mailer = createAppMailer(createMailProvider(config), config.MAIL_FROM)

export default {
  worker: createWorker({
    listeners: createNotifications({ repositories, mailer }),
    logger,
    // The attempts of the BullMQ jobs (3 by default): the last failure is
    // reported as final, for alerts.
    maxAttempts: 3,
    onFailure: (error, { envelope, attempt, final }) => {
      if (final) {
        logger.error(
          { error, listener: envelope?.listener, id: envelope?.id, attempt },
          'Notification given up: the job stays in the failed jobs'
        )
      }
    },
  }),
  source: createBullmqSource({
    name: config.EVENTS_QUEUE,
    connection: redisConnection(config.REDIS_URL),
    concurrency: 5,
  }),
  logger,
}