diff --git a/README.md b/README.md index 87e7904..521e5c7 100644 --- a/README.md +++ b/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. diff --git a/src/app.js b/src/app.js index f3b0771..bd5bcbc 100644 --- a/src/app.js +++ b/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, }) } diff --git a/src/attachments-audit.js b/src/attachments-audit.js new file mode 100644 index 0000000..4826b8d --- /dev/null +++ b/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} 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>[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} + */ +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} + */ +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[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 +} diff --git a/src/routes.js b/src/routes.js index 01f202d..5ac2767 100644 --- a/src/routes.js +++ b/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} */ export default async function routes( fastify, - { serverBearerToken, serverName, allowedProjects = 1 }, + { serverBearerToken, serverName, allowedProjects = 1, auditMaxAttachments }, ) { /** @type {Set | 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 diff --git a/src/schemas.js b/src/schemas.js index d7d4d75..1b3203f 100644 --- a/src/schemas.js +++ b/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, diff --git a/src/server.js b/src/server.js index cae373d..5bea5a6 100644 --- a/src/server.js +++ b/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} Env */ @@ -90,6 +95,7 @@ fastify.register(comapeoServer, { dbFolder, projectMigrationsFolder, clientMigrationsFolder, + auditMaxAttachments: config.AUDIT_MAX_ATTACHMENTS, }) fastify.get('/healthcheck', async () => {}) diff --git a/test/attachments-audit-endpoint.js b/test/attachments-audit-endpoint.js new file mode 100644 index 0000000..b38d739 --- /dev/null +++ b/test/attachments-audit-endpoint.js @@ -0,0 +1,99 @@ +import assert from 'node:assert/strict' +import test from 'node:test' + +import { + BEARER_TOKEN, + FIXTURE_AUDIO_PATH, + FIXTURE_IMAGE_ORIGINAL_PATH, + FIXTURE_IMAGE_PREVIEW_PATH, + FIXTURE_IMAGE_THUMBNAIL_PATH, + blobToAttachment, + createPartialPhotoBlob, + createTestServer, + generateObservation, +} from './test-helpers.js' + +test('returns a 401 if no auth is provided', async (t) => { + const server = createTestServer(t) + await server.listen() + + const projectPublicId = await server.comapeo.createProject({ + name: 'project', + }) + + const response = await server.inject({ + method: 'GET', + url: `/projects/${projectPublicId}/attachments/audit`, + }) + assert.equal(response.statusCode, 401) + assert.equal(response.json().error.code, 'UNAUTHORIZED') +}) + +test('counts every attachment variant and reports the partly synced one', async (t) => { + const server = createTestServer(t) + await server.listen() + + await server.comapeo.setDeviceInfo({ + deviceType: 'selfHostedServer', + name: 'test server', + }) + const projectPublicId = await server.comapeo.createProject({ + name: 'project', + }) + const project = await server.comapeo.getProject(projectPublicId) + + const photo = await project.$blobs.create( + { + original: FIXTURE_IMAGE_ORIGINAL_PATH, + preview: FIXTURE_IMAGE_PREVIEW_PATH, + thumbnail: FIXTURE_IMAGE_THUMBNAIL_PATH, + }, + { mimeType: 'image/jpeg' }, + ) + const audio = await project.$blobs.create( + { original: FIXTURE_AUDIO_PATH }, + { mimeType: 'audio/mpeg' }, + ) + const partialPhoto = await createPartialPhotoBlob(t, project) + await project.observation.create({ + ...generateObservation(), + attachments: [blobToAttachment(photo), blobToAttachment(audio)], + }) + await project.observation.create({ + ...generateObservation(), + attachments: [blobToAttachment(partialPhoto)], + }) + + const response = await server.inject({ + method: 'GET', + url: `/projects/${projectPublicId}/attachments/audit`, + headers: { Authorization: 'Bearer ' + BEARER_TOKEN }, + }) + assert.equal(response.statusCode, 200) + /** @type {import('../src/attachments-audit.js').AttachmentAudit} */ + const { failures, devices, ...counts } = response.json().data + // 3 attachments: two photos with 3 variants each, one audio with 1 + assert.deepEqual(counts, { + observations: 2, + attachments: 3, + audited: 3, + ok: 6, + missing: 0, + incomplete: 1, + error: 0, + }) + assert.deepEqual( + failures.map(({ detail: _detail, ...failure }) => failure), + [ + { + driveDiscoveryId: partialPhoto.driveId, + createdBy: server.comapeo.deviceId, + type: 'photo', + name: partialPhoto.name, + variant: 'original', + status: 'incomplete', + }, + ], + ) + assert.deepEqual(devices, { [server.comapeo.deviceId]: 'test server' }) +}) diff --git a/test/attachments-endpoint.js b/test/attachments-endpoint.js index 04b0f87..1aaa66e 100644 --- a/test/attachments-endpoint.js +++ b/test/attachments-endpoint.js @@ -1,17 +1,11 @@ import assert from 'node:assert/strict' -import { randomBytes } from 'node:crypto' -import * as fs from 'node:fs/promises' -import * as os from 'node:os' -import * as path from 'node:path' import test from 'node:test' -import { BEARER_TOKEN, createTestServer } from './test-helpers.js' - -// Test-only internals, not in core's package exports or public types -const { kBlobStore, kCoreManager } = await import( - new URL('../node_modules/@comapeo/core/src/mapeo-project.js', import.meta.url) - .href -) +import { + BEARER_TOKEN, + createPartialPhotoBlob, + createTestServer, +} from './test-helpers.js' test('returns 502 for an attachment that is only partly synced', async (t) => { const server = createTestServer(t) @@ -21,30 +15,7 @@ test('returns 502 for an attachment that is only partly synced', async (t) => { name: 'project', }) const project = await server.comapeo.getProject(projectPublicId) - const internals = /** @type {any} */ (project) - - const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'comapeo-cloud-test-')) - t.after(() => fs.rm(dir, { recursive: true, force: true })) - const filePath = path.join(dir, 'original.jpg') - // Three hypercore blocks (64 KiB each), so a partial blob is possible - await fs.writeFile(filePath, randomBytes(3 * 64 * 1024)) - - const blob = await project.$blobs.create( - { original: filePath }, - { mimeType: 'image/jpeg' }, - ) - const entry = await internals[kBlobStore].entry({ - driveId: blob.driveId, - type: 'photo', - variant: 'original', - name: blob.name, - }) - assert(entry, 'test setup: blob entry exists') - const { blockOffset, blockLength } = entry.value.blob - // Keep block 0 and drop the rest, like an original mid-way through sync - await internals[kCoreManager] - .getWriterCore('blob') - .core.clear(blockOffset + 1, blockOffset + blockLength) + const blob = await createPartialPhotoBlob(t, project) const response = await server.inject({ method: 'GET', diff --git a/test/observations-endpoint.js b/test/observations-endpoint.js index 09fb67f..4fa7714 100644 --- a/test/observations-endpoint.js +++ b/test/observations-endpoint.js @@ -8,6 +8,11 @@ import test from 'node:test' import { BEARER_TOKEN, + FIXTURE_AUDIO_PATH, + FIXTURE_IMAGE_ORIGINAL_PATH, + FIXTURE_IMAGE_PREVIEW_PATH, + FIXTURE_IMAGE_THUMBNAIL_PATH, + blobToAttachment, createTestServer, generateObservation, getManagerOptions, @@ -19,15 +24,6 @@ import { /** @import { ObservationValue } from '@comapeo/core/schema.js'*/ /** @import { FastifyInstance } from 'fastify' */ -const FIXTURES_ROOT = new URL('./fixtures/', import.meta.url) -const FIXTURE_IMAGE_ORIGINAL_PATH = new URL('original.jpg', FIXTURES_ROOT) - .pathname -const FIXTURE_IMAGE_PREVIEW_PATH = new URL('preview.jpg', FIXTURES_ROOT) - .pathname -const FIXTURE_IMAGE_THUMBNAIL_PATH = new URL('thumbnail.jpg', FIXTURES_ROOT) - .pathname -const FIXTURE_AUDIO_PATH = new URL('audio.mp3', FIXTURES_ROOT).pathname - test('returns a 401 if no auth is provided', async (t) => { const server = createTestServer(t) @@ -170,23 +166,6 @@ test('returning observations with fetchable attachments', async (t) => { ) }) -/** - * @param {object} blob - * @param {string} blob.driveId - * @param {'photo' | 'audio' | 'video'} blob.type - * @param {string} blob.name - * @param {string} blob.hash - */ -function blobToAttachment(blob) { - return { - driveDiscoveryId: blob.driveId, - type: blob.type, - name: blob.name, - hash: blob.hash, - external: false, - } -} - /** * @param {object} options * @param {FastifyInstance} options.server diff --git a/test/test-helpers.js b/test/test-helpers.js index 4da62d7..5239aa3 100644 --- a/test/test-helpers.js +++ b/test/test-helpers.js @@ -9,6 +9,9 @@ import RAM from 'random-access-memory' import assert from 'node:assert/strict' import { randomBytes } from 'node:crypto' +import * as fs from 'node:fs/promises' +import * as os from 'node:os' +import * as path from 'node:path' import { setTimeout as delay } from 'node:timers/promises' import { fileURLToPath } from 'node:url' @@ -20,8 +23,27 @@ import comapeoServer from '../src/app.js' /** @import { ServerOptions } from '../src/app.js' */ /** @import { MapeoProject } from '@comapeo/core' */ +// Test-only internals, not in core's package exports or public types +const { kBlobStore, kCoreManager } = await import( + new URL('../node_modules/@comapeo/core/src/mapeo-project.js', import.meta.url) + .href +) + export const BEARER_TOKEN = Buffer.from('swordfish').toString('base64') +const FIXTURES_ROOT = new URL('./fixtures/', import.meta.url) +export const FIXTURE_IMAGE_ORIGINAL_PATH = new URL( + 'original.jpg', + FIXTURES_ROOT, +).pathname +export const FIXTURE_IMAGE_PREVIEW_PATH = new URL('preview.jpg', FIXTURES_ROOT) + .pathname +export const FIXTURE_IMAGE_THUMBNAIL_PATH = new URL( + 'thumbnail.jpg', + FIXTURES_ROOT, +).pathname +export const FIXTURE_AUDIO_PATH = new URL('audio.mp3', FIXTURES_ROOT).pathname + const TEST_SERVER_DEFAULTS = { serverName: 'test server', serverBearerToken: BEARER_TOKEN, @@ -204,3 +226,58 @@ export function generateTrack() { assert(trackDoc) return valueOf(trackDoc) } + +/** + * @param {object} blob + * @param {string} blob.driveId + * @param {'photo' | 'audio' | 'video'} blob.type + * @param {string} blob.name + * @param {string} blob.hash + */ +export function blobToAttachment(blob) { + return { + driveDiscoveryId: blob.driveId, + type: blob.type, + name: blob.name, + hash: blob.hash, + external: false, + } +} + +/** + * Create a photo whose original has only its first block, like one caught + * mid-way through sync + * @param {TestContext} t + * @param {MapeoProject} project + */ +export async function createPartialPhotoBlob(t, project) { + const internals = /** @type {any} */ (project) + + const dir = await fs.mkdtemp(path.join(os.tmpdir(), 'comapeo-cloud-test-')) + t.after(() => fs.rm(dir, { recursive: true, force: true })) + const filePath = path.join(dir, 'original.jpg') + // Three hypercore blocks (64 KiB each), so a partial blob is possible + await fs.writeFile(filePath, randomBytes(3 * 64 * 1024)) + + const blob = await project.$blobs.create( + { + original: filePath, + preview: FIXTURE_IMAGE_PREVIEW_PATH, + thumbnail: FIXTURE_IMAGE_THUMBNAIL_PATH, + }, + { mimeType: 'image/jpeg' }, + ) + const entry = await internals[kBlobStore].entry({ + driveId: blob.driveId, + type: 'photo', + variant: 'original', + name: blob.name, + }) + assert(entry, 'test setup: blob entry exists') + const { blockOffset, blockLength } = entry.value.blob + await internals[kCoreManager] + .getWriterCore('blob') + .core.clear(blockOffset + 1, blockOffset + blockLength) + + return blob +}