feat(attachments): add an attachment audit endpoint

README.md

@@ -22,13 +22,14 @@ Or use the [`Dockerfile`](./Dockerfile) to build a Docker image.
 
 Server configuration is done using environment variables. The following environment variables are available:
 
-| Environment Variable  | Required | Description                                                          | Default Value    |
-| --------------------- | -------- | -------------------------------------------------------------------- | ---------------- |
-| `SERVER_BEARER_TOKEN` | Yes      | Token for authenticating API requests. Should be large random string |                  |
-| `PORT`                | No       | Port on which the server runs                                        | `8080`           |
-| `SERVER_NAME`         | No       | Friendly server name, seen by users when adding server               | `CoMapeo Server` |
-| `ALLOWED_PROJECTS`    | No       | Number of projects allowed to register with the server               | `1`              |
-| `STORAGE_DIR`         | No       | Path for storing app & project data                                  | `$CWD/data`      |
+| Environment Variable    | Required | Description                                                           | Default Value    |
+| ----------------------- | -------- | --------------------------------------------------------------------- | ---------------- |
+| `SERVER_BEARER_TOKEN`   | Yes      | Token for authenticating API requests. Should be large random string  |                  |
+| `PORT`                  | No       | Port on which the server runs                                         | `8080`           |
+| `SERVER_NAME`           | No       | Friendly server name, seen by users when adding server                | `CoMapeo Server` |
+| `ALLOWED_PROJECTS`      | No       | Number of projects allowed to register with the server                | `1`              |
+| `STORAGE_DIR`           | No       | Path for storing app & project data                                   | `$CWD/data`      |
+| `AUDIT_MAX_ATTACHMENTS` | No       | Attachments checked per project by the attachment audit, newest first | `200`            |
 
 If you are using Nginx to act as a reverse proxy for your CoMapeo Cloud server, ensure your proxy headers are configured to support WebSockets.
 

src/app.js

@@ -29,6 +29,7 @@ async function comapeoServer(
     serverName,
     allowedHosts,
     allowedProjects,
+    auditMaxAttachments,
     ...comapeoPluginOpts
   },
 ) {
@@ -41,6 +42,7 @@ async function comapeoServer(
     serverBearerToken,
     serverName,
     allowedProjects,
+    auditMaxAttachments,
   })
 }
 

src/attachments-audit.js

@@ -0,0 +1,175 @@
+/** @import { MapeoProject } from '@comapeo/core' */
+/** @import { Static } from '@sinclair/typebox' */
+/** @import { attachmentAudit } from './schemas.js' */
+
+/** @typedef {Static<typeof attachmentAudit>} AttachmentAudit */
+/** @typedef {AttachmentAudit['failures'][number]} AttachmentFailure */
+/** @typedef {AttachmentFailure['status'] | 'ok'} AttachmentStatus */
+/**
+ * @typedef {object} AttachmentCheck
+ * @prop {AttachmentStatus} status
+ * @prop {string} detail one line for humans: upstream status and code, or why the read failed
+ * @prop {string} [upstreamCode] error code sent by core's blob server, when it sent one
+ */
+
+const VARIANTS = /** @type {const} */ ({
+  photo: ['original', 'preview', 'thumbnail'],
+  audio: ['original'],
+})
+
+/** @typedef {Awaited<ReturnType<MapeoProject['observation']['getMany']>>[number]['attachments'][number]} Attachment */
+
+/**
+ * @param {Attachment} attachment
+ * @returns {attachment is Attachment & { type: keyof typeof VARIANTS }}
+ */
+const isSupportedAttachment = (attachment) =>
+  Object.hasOwn(VARIANTS, attachment.type)
+
+/**
+ * Classify a non-200 response from core's blob server
+ * @param {Response} response
+ * @param {string} body
+ * @returns {AttachmentCheck}
+ */
+export function upstreamFailure(response, body) {
+  /** @type {string | undefined} */
+  let upstreamCode
+  try {
+    const parsed = JSON.parse(body)
+    if (typeof parsed?.code === 'string') upstreamCode = parsed.code
+  } catch {
+    // not JSON
+  }
+  return {
+    status: response.status === 404 ? 'missing' : 'error',
+    detail: `${response.status} ${upstreamCode || response.statusText}`.trim(),
+    upstreamCode,
+  }
+}
+
+/**
+ * @param {unknown} err
+ * @returns {AttachmentCheck}
+ */
+export function fetchFailure(err) {
+  if (!(err instanceof Error)) {
+    return { status: 'incomplete', detail: String(err) }
+  }
+  // undici's message is always 'fetch failed'; the reason is in the cause
+  const cause = /** @type {{ code?: string, message?: string } | undefined} */ (
+    err.cause
+  )
+  const reason = cause?.code ?? cause?.message
+  return {
+    status: 'incomplete',
+    detail: reason ? `${err.message}: ${reason}` : err.message,
+  }
+}
+
+/**
+ * Fetch a blob from core's blob server and read it to the end
+ * @param {string} blobUrl
+ * @returns {Promise<AttachmentCheck>}
+ */
+export async function checkBlob(blobUrl) {
+  let response
+  try {
+    response = await fetch(blobUrl, { signal: AbortSignal.timeout(60_000) })
+  } catch (err) {
+    return fetchFailure(err)
+  }
+  if (response.status !== 200) {
+    return upstreamFailure(response, await response.text())
+  }
+  let bytes = 0
+  try {
+    if (response.body) {
+      for await (const chunk of response.body) bytes += chunk.length
+    }
+  } catch (err) {
+    return {
+      status: 'incomplete',
+      detail: `${fetchFailure(err).detail} after ${bytes} bytes`,
+    }
+  }
+  return { status: 'ok', detail: `${bytes} bytes` }
+}
+
+/**
+ * Check every variant of every photo and audio attachment in a project. Each
+ * failure carries the device that created the observation, since that is the
+ * device that has to sync again, and `devices` names those devices.
+ * @param {MapeoProject} project
+ * @param {object} [options]
+ * @param {number} [options.maxAttachments] newest observations first
+ * @returns {Promise<AttachmentAudit>}
+ */
+export async function auditAttachments(
+  project,
+  { maxAttachments = Infinity } = {},
+) {
+  const observations = await project.observation.getMany()
+  observations.sort((a, b) => (a.createdAt < b.createdAt ? 1 : -1))
+
+  /** @type {AttachmentAudit} */
+  const result = {
+    observations: observations.length,
+    attachments: 0,
+    audited: 0,
+    ok: 0,
+    missing: 0,
+    incomplete: 0,
+    error: 0,
+    failures: [],
+    devices: {},
+  }
+
+  const attachments = observations.flatMap((obs) =>
+    obs.attachments
+      .filter(isSupportedAttachment)
+      .map((attachment) => ({ attachment, createdBy: obs.createdBy })),
+  )
+  result.attachments = attachments.length
+
+  for (const { attachment, createdBy } of attachments.slice(
+    0,
+    maxAttachments,
+  )) {
+    const { driveDiscoveryId, type, name } = attachment
+    result.audited++
+    for (const variant of VARIANTS[type]) {
+      const blobUrl = await project.$blobs.getUrl(
+        /** @type {Parameters<typeof project.$blobs.getUrl>[0]} */ ({
+          driveId: driveDiscoveryId,
+          type,
+          variant,
+          name,
+        }),
+      )
+      const check = await checkBlob(blobUrl)
+      result[check.status]++
+      if (check.status !== 'ok') {
+        result.failures.push({
+          driveDiscoveryId,
+          createdBy,
+          type,
+          name,
+          variant,
+          status: check.status,
+          detail: check.detail,
+        })
+      }
+    }
+  }
+
+  if (result.failures.length > 0) {
+    const members = await project.$member.getMany({ includeLeft: true })
+    const createdBy = new Set(result.failures.map((f) => f.createdBy))
+    for (const { deviceId, name } of members) {
+      if (name && createdBy.has(deviceId)) result.devices[deviceId] = name
+    }
+  }
+
+  return result
+}

src/routes.js

@@ -7,6 +7,7 @@ import assert from 'node:assert/strict'
 import * as fs from 'node:fs'
 import { STATUS_CODES } from 'node:http'
 
+import { auditAttachments } from './attachments-audit.js'
 import { Field as fieldSchema } from './datatypes/field.js'
 import { Observation as observationSchema } from './datatypes/observation.js'
 import { Preset as presetSchema } from './datatypes/preset.js'
@@ -57,12 +58,13 @@ const HOP_BY_HOP_HEADERS = new Set([
  * @prop {string} serverBearerToken
  * @prop {string} serverName
  * @prop {undefined | number | string[]} [allowedProjects=1]
+ * @prop {number} [auditMaxAttachments] cap on attachments checked per audit
  */
 
 /** @type {FastifyPluginAsync<RouteOptions, RawServerDefault, TypeBoxTypeProvider>} */
 export default async function routes(
   fastify,
-  { serverBearerToken, serverName, allowedProjects = 1 },
+  { serverBearerToken, serverName, allowedProjects = 1, auditMaxAttachments },
 ) {
   /** @type {Set<string> | number} */
   const allowedProjectsSetOrNumber = Array.isArray(allowedProjects)
@@ -547,6 +549,37 @@ export default async function routes(
     },
   )
 
+  fastify.get(
+    '/projects/:projectPublicId/attachments/audit',
+    {
+      schema: {
+        params: Type.Object({
+          projectPublicId: BASE32_STRING_32_BYTES,
+        }),
+        response: {
+          200: Type.Object({
+            data: schemas.attachmentAudit,
+          }),
+          '4xx': schemas.errorResponse,
+        },
+      },
+      async preHandler(req) {
+        verifyBearerAuth(req)
+        await ensureProjectExists(this, req)
+      },
+    },
+    /**
+     * @this {FastifyInstance}
+     */
+    async function (req) {
+      const project = await this.comapeo.getProject(req.params.projectPublicId)
+      const data = await auditAttachments(project, {
+        maxAttachments: auditMaxAttachments,
+      })
+      return { data }
+    },
+  )
+
   /**
    * @template {TSchema} Schema
    * @template {"track"|"observation"|"preset"|"field"} TDataType

src/schemas.js

@@ -14,6 +14,50 @@ export const errorResponse = Type.Object({
   }),
 })
 
+const attachmentStatus = Type.Union([
+  Type.Literal('ok'),
+  Type.Literal('missing'),
+  Type.Literal('incomplete'),
+  Type.Literal('error'),
+])
+
+export const attachmentAudit = Type.Object({
+  observations: Type.Integer(),
+  attachments: Type.Integer({
+    description: 'photo and audio attachments found on observations',
+  }),
+  audited: Type.Integer({
+    description: 'attachments whose variants were read',
+  }),
+  ok: Type.Integer({ description: 'variants read to the end' }),
+  missing: Type.Integer({ description: 'variants not found (404)' }),
+  incomplete: Type.Integer({
+    description: 'variants that reset or ended before their last byte',
+  }),
+  error: Type.Integer({ description: 'variants that failed in any other way' }),
+  failures: Type.Array(
+    Type.Object({
+      driveDiscoveryId: Type.String(),
+      createdBy: Type.String({
+        description: 'device id of the device that created the observation',
+      }),
+      type: Type.Union([Type.Literal('photo'), Type.Literal('audio')]),
+      name: Type.String(),
+      variant: Type.Union([
+        Type.Literal('original'),
+        Type.Literal('preview'),
+        Type.Literal('thumbnail'),
+      ]),
+      status: attachmentStatus,
+      detail: Type.String(),
+    }),
+  ),
+  devices: Type.Record(Type.String(), Type.String(), {
+    description:
+      'name of each device that appears in failures, by device id, where the project knows it',
+  }),
+})
+
 export const projectToAdd = Type.Object({
   projectName: Type.String({ minLength: 1 }),
   projectKey: HEX_STRING_32_BYTES,

src/server.js

@@ -33,6 +33,11 @@ const schema = Type.Object({
       description: 'number of projects allowed to join the server',
     }),
   ),
+  AUDIT_MAX_ATTACHMENTS: Type.Integer({
+    minimum: 1,
+    default: 200,
+    description: 'attachments checked per project per audit, newest first',
+  }),
 })
 
 /** @typedef {import('@sinclair/typebox').Static<typeof schema>} Env */
@@ -90,6 +95,7 @@ fastify.register(comapeoServer, {
   dbFolder,
   projectMigrationsFolder,
   clientMigrationsFolder,
+  auditMaxAttachments: config.AUDIT_MAX_ATTACHMENTS,
 })
 
 fastify.get('/healthcheck', async () => {})