Support S3-compatible blob storage

.github/workflows/s3-e2e.yml

@@ -0,0 +1,54 @@
+name: S3 E2E Tests
+
+on: push
+
+env:
+  LOG_LEVEL: DEBUG
+
+jobs:
+  s3-e2e:
+    timeout-minutes: 15
+    # TODO should we use the same container as circle & central?
+    runs-on: ubuntu-latest
+    services:
+      # see: https://docs.github.com/en/enterprise-server@3.5/actions/using-containerized-services/creating-postgresql-service-containers
+      postgres:
+        image: postgres:14.10
+        env:
+          POSTGRES_PASSWORD: odktest
+        ports:
+          - 5432:5432
+        # Set health checks to wait until postgres has started
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+      minio:
+        # see: https://github.com/minio/minio/discussions/16099
+        image: minio/minio:edge-cicd
+        env:
+          MINIO_ROOT_USER: odk-central-dev
+          MINIO_ROOT_PASSWORD: topSecret123
+        ports:
+          - 9000:9000
+        options: >-
+          --health-cmd "curl -s http://localhost:9000/minio/health/live"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    steps:
+    - uses: actions/checkout@v4
+    - name: Use Node.js 20
+      uses: actions/setup-node@v4
+      with:
+        node-version: 20.10.0
+        cache: 'npm'
+    - run: npm ci --legacy-peer-deps
+    - run: node lib/bin/create-docker-databases.js
+    - name: E2E Test
+      timeout-minutes: 10
+      run: ./test/e2e/s3/ci
+    - name: Backend Logs
+      if: always()
+      run: "! [[ -f ./backend.log ]] || cat ./backend.log"

.github/workflows/s3-integration.yml

@@ -0,0 +1,46 @@
+name: S3 integration tests
+
+on: push
+
+jobs:
+  s3-integration-test:
+    timeout-minutes: 4
+    # TODO should we use the same container as circle & central?
+    runs-on: ubuntu-latest
+    services:
+      # see: https://docs.github.com/en/enterprise-server@3.5/actions/using-containerized-services/creating-postgresql-service-containers
+      postgres:
+        image: postgres:14.10
+        env:
+          POSTGRES_PASSWORD: odktest
+        ports:
+          - 5432:5432
+        # Set health checks to wait until postgres has started
+        options: >-
+          --health-cmd pg_isready
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+      minio:
+        # see: https://github.com/minio/minio/discussions/16099
+        image: minio/minio:edge-cicd
+        env:
+          MINIO_ROOT_USER: odk-central-dev
+          MINIO_ROOT_PASSWORD: topSecret123
+        ports:
+          - 9000:9000
+        options: >-
+          --health-cmd "curl -s http://localhost:9000/minio/health/live"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+    steps:
+    - uses: actions/checkout@v4
+    - name: Use Node.js 20
+      uses: actions/setup-node@v4
+      with:
+        node-version: 20.10.0
+        cache: 'npm'
+    - run: npm ci --legacy-peer-deps
+    - run: node lib/bin/create-docker-databases.js
+    - run: make test-s3-integration

Makefile

@@ -8,6 +8,10 @@ node_modules: package.json
 test-oidc-integration: node_version
 	TEST_AUTH=oidc NODE_CONFIG_ENV=oidc-integration-test make test-integration
 
+.PHONY: test-s3-integration
+test-s3-integration: fake-s3-accounts
+	TEST_S3=true NODE_CONFIG_ENV=s3-dev make test-integration
+
 .PHONY: test-oidc-e2e
 test-oidc-e2e: node_version
 	cd test/e2e/oidc && \
@@ -19,6 +23,14 @@ test-oidc-e2e: node_version
 dev-oidc: base
 	NODE_CONFIG_ENV=oidc-development npx nodemon --watch lib --watch config lib/bin/run-server.js
 
+.PHONY: fake-s3-accounts
+fake-s3-accounts: node_version
+	NODE_CONFIG_ENV=s3-dev node lib/bin/s3-create-bucket.js
+
+.PHONY: dev-s3
+dev-s3: fake-s3-accounts base
+	NODE_CONFIG_ENV=s3-dev npx nodemon --watch lib --watch config lib/bin/run-server.js
+
 .PHONY: fake-oidc-server
 fake-oidc-server:
 	cd test/e2e/oidc/fake-oidc-server && \
@@ -31,6 +43,17 @@ fake-oidc-server-ci:
 	npm clean-install && \
 	FAKE_OIDC_ROOT_URL=http://localhost:9898 node index.js
 
+.PHONY: fake-s3-server
+fake-s3-server:
+	# run an ephemeral, s3-compatible local store
+	# default admin credentials: minioadmin:minioadmin
+	# see: https://hub.docker.com/r/minio/minio/
+	docker run --rm \
+		-p 9000:9000 -p 9001:9001 \
+		-e MINIO_ROOT_USER=odk-central-dev \
+		-e MINIO_ROOT_PASSWORD=topSecret123 \
+		minio/minio server /data --console-address ":9001"
+
 .PHONY: node_version
 node_version: node_modules
 	node lib/bin/enforce-node-version.js

config/s3-dev.json

@@ -0,0 +1,11 @@
+{
+  "default": {
+    "s3blobStore": {
+      "server": "http://localhost:9000",
+      "accessKey": "odk-central-dev",
+      "secretKey": "topSecret123",
+      "bucketName": "odk-central-bucket",
+      "requestTimeout": 60000
+    }
+  }
+}

lib/bin/run-server.js

@@ -63,6 +63,9 @@ const server = service.listen(config.get('default.server.port'), () => {
 const { workerQueue } = require('../worker/worker');
 workerQueue(container).loops(4);
 
+const { s3worker } = require('../util/s3');
+s3worker.start(container);
+
 
 ////////////////////////////////////////////////////////////////////////////////
 // CLEANUP

lib/bin/s3-create-bucket.js

@@ -0,0 +1,39 @@
+// Copyright 2024 ODK Central Developers
+// See the NOTICE file at the top-level directory of this distribution and at
+// https://github.com/getodk/central-backend/blob/master/NOTICE.
+// This file is part of ODK Central. It is subject to the license terms in
+// the LICENSE file found in the top-level directory of this distribution and at
+// https://www.apache.org/licenses/LICENSE-2.0. No part of ODK Central,
+// including this file, may be copied, modified, propagated, or distributed
+// except according to the terms contained in the LICENSE file.
+
+const Minio = require('minio');
+
+const { server, bucketName, accessKey, secretKey } = require('config').get('default').s3blobStore;
+
+const minioClient = (() => {
+  const url = new URL(server);
+  const useSSL = url.protocol === 'https:';
+  const endPoint = (url.hostname + url.pathname).replace(/\/$/, '');
+  const port = parseInt(url.port, 10);
+
+  return new Minio.Client({ endPoint, port, useSSL, accessKey, secretKey });
+})();
+
+const log = (...args) => console.log(__filename, ...args);
+
+minioClient.bucketExists(bucketName)
+  .then(exists => {
+    if (exists) {
+      log('Bucket already exists.');
+      return;
+    }
+
+    log('Creating bucket:', bucketName);
+    return minioClient.makeBucket(bucketName)
+      .then(() => log('Bucket created OK.'));
+  })
+  .catch(err => {
+    log('ERROR CREATING MINIO BUCKET:', err);
+    process.exit(1);
+  });

lib/bin/s3.js

@@ -0,0 +1,59 @@
+// Copyright 2024 ODK Central Developers
+// See the NOTICE file at the top-level directory of this distribution and at
+// https://github.com/getodk/central-backend/blob/master/NOTICE.
+// This file is part of ODK Central. It is subject to the license terms in
+// the LICENSE file found in the top-level directory of this distribution and at
+// https://www.apache.org/licenses/LICENSE-2.0. No part of ODK Central,
+// including this file, may be copied, modified, propagated, or distributed
+// except according to the terms contained in the LICENSE file.
+
+/* eslint-disable no-use-before-define */
+
+const config = require('config');
+const { program } = require('commander');
+const { sql } = require('slonik');
+
+const { slonikPool } = require('../external/slonik');
+const { exhaustBlobs } = require('../util/s3');
+
+program.command('count-failed-uploads').action(countFailed);
+program.command('reset-failed-as-pending').action(setFailedToPending);
+program.command('upload-pending').action(uploadPending);
+program.parse();
+
+function countFailed() {
+  withDb(async (db) => console.log('Failed uploads:', await getFailedCount(db)));
+}
+
+function setFailedToPending() {
+  withDb(async (db) => {
+    const count = await db.oneFirst(sql`
+      WITH updated AS (
+        UPDATE blobs SET s3_status='pending' WHERE s3_status='failed' RETURNING 1
+      )
+      SELECT COUNT(*) FROM updated
+    `);
+    console.log(count, 'blobs marked for re-uploading.');
+  });
+}
+
+function uploadPending() {
+  withDb(async (db) => {
+    console.log('Uploading', await getFailedCount(db), 'blobs...');
+    await exhaustBlobs({ db });
+    console.log('Upload completed.');
+  });
+}
+
+function getFailedCount(db) {
+  return db.oneFirst(sql`SELECT COUNT(*) FROM blobs WHERE s3_status='failed'`);
+}
+
+async function withDb(fn) {
+  const db = slonikPool(config.get('default.database'));
+  try {
+    await fn(db);
+  } finally {
+    await db.end();
+  }
+}

lib/data/attachments.js

@@ -13,6 +13,7 @@ const { compose, identity } = require('ramda');
 const { Writable, pipeline } = require('stream');
 const { rejectIfError } = require('../util/promise');
 const { zipPart } = require('../util/zip');
+const s3 = require('../util/s3');
 const sanitize = require('sanitize-filename');
 
 // encrypted files have a .enc extension that needs to be stripped. we will only
@@ -33,14 +34,24 @@ const streamAttachments = (inStream, decryptor) => {
     write(x, _, done) {
       const att = x.row;
 
-      // this sanitization means that two filenames could end up identical.
-      // luckily, this is not actually illegal in the zip spec; two files can live at precisely
-      // the same location, and the conflict is dealt with interactively by the unzipping client.
-      const content = (att.localKey == null)
-        ? att.content
-        : decryptor(att.content, att.keyId, att.localKey, att.instanceId, att.index);
+      const appendContent = (rawContent) => {
+        // this sanitization means that two filenames could end up identical.
+        // luckily, this is not actually illegal in the zip spec; two files can live at precisely
+        // the same location, and the conflict is dealt with interactively by the unzipping client.
+        const content = (att.localKey == null)
+          ? rawContent
+          : decryptor(rawContent, att.keyId, att.localKey, att.instanceId, att.index);
 
-      archive.append(content, { name: join('media', processName(att.name)) }, done);
+        archive.append(content, { name: join('media', processName(att.name)) }, done);
+      };
+
+      if (att.s3_status === 'uploaded') {
+        s3.getContentFor(att)
+          .then(appendContent)
+          .catch(err => { this.destroy(err); });
+      } else {
+        appendContent(att.content);
+      }
     },
     final(done) {
       archive.finalize();

lib/data/briefcase.js

@@ -16,6 +16,7 @@ const { SchemaStack } = require('./schema');
 const { rejectIfError } = require('../util/promise');
 const { PartialPipe } = require('../util/stream');
 const { zipPart } = require('../util/zip');
+const s3 = require('../util/s3');
 
 /* eslint-disable no-multi-spaces */
 
@@ -293,27 +294,40 @@ const streamBriefcaseCsvs = (inStream, inFields, xmlFormId, selectValues, decryp
         // the decryptor will return null if it does not have a decryption key for the
         // record. this is okay; we pass the null through and processRow deals with it.
         const { encryption } = submission.aux;
-        const xml =
-          (submission.def.localKey == null) ? submission.xml :
-          (encryption.encHasData === false) ? 'missing' : // eslint-disable-line indent
-          decryptor(encryption.encData, encryption.encKeyId, submission.def.localKey, submission.instanceId, encryption.encIndex); // eslint-disable-line indent
-
-        // if something about the xml didn't work so well, we can figure out what
-        // to say and bail out early.
-        const status =
-          (xml === 'missing') ? 'missing encrypted form data' :
-          (xml === null) ? 'not decrypted' : null; // eslint-disable-line indent
-        if (status != null) {
-          const result = new Array(rootHeader.length);
-          writeMetadata(result, rootMeta, submission, submission.aux.submitter, submission.aux.exports.formVersion, submission.aux.attachment, status);
-          return done(null, result);
+
+        const decryptXml = (data) => decryptor(data, encryption.encKeyId, submission.def.localKey, submission.instanceId, encryption.encIndex);
+        const processXml = (xml) => {
+          try {
+            // if something about the xml didn't work so well, we can figure out what
+            // to say and bail out early.
+            const status =
+              (xml === 'missing') ? 'missing encrypted form data' :
+              (xml === null) ? 'not decrypted' : null; // eslint-disable-line indent
+            if (status != null) {
+              const result = new Array(rootHeader.length);
+              writeMetadata(result, rootMeta, submission, submission.aux.submitter, submission.aux.exports.formVersion, submission.aux.attachment, status);
+              return done(null, result);
+            }
+
+            // write the root row we get back from parsing the xml.
+            processRow(xml, submission.instanceId, fields, rootHeader, selectValues).then((result) => {
+              writeMetadata(result, rootMeta, submission, submission.aux.submitter, submission.aux.exports.formVersion, submission.aux.attachment);
+              done(null, result);
+            }, done); // pass through errors.
+          } catch (ex) { done(ex); }
+        };
+
+        if (submission.def.localKey == null) processXml(submission.xml);
+        else if (encryption.encHasData === false) processXml('missing');
+        else if (encryption.encS3Status === 'uploaded') {
+          s3.getContentFor({ sha: encryption.encSha, md5: encryption.encMd5 })
+            .then(decryptXml)
+            .then(processXml)
+            .catch(done);
+        } else {
+          processXml(decryptXml(encryption.encData));
         }
 
-        // write the root row we get back from parsing the xml.
-        processRow(xml, submission.instanceId, fields, rootHeader, selectValues).then((result) => {
-          writeMetadata(result, rootMeta, submission, submission.aux.submitter, submission.aux.exports.formVersion, submission.aux.attachment);
-          done(null, result);
-        }, done); // pass through errors.
       } catch (ex) { done(ex); }
     }, flush(done) {
       if (rootHeaderSent === false) this.push(rootHeader);

lib/model/frames/blob.js

@@ -16,7 +16,7 @@ const { digestWith, md5sum, shasum } = require('../../util/crypto');
 const { pipethroughAndBuffer } = require('../../util/stream');
 
 
-class Blob extends Frame.define(table('blobs'), 'id', 'sha', 'content', 'contentType', 'md5') {
+class Blob extends Frame.define(table('blobs'), 'id', 'sha', 'content', 'contentType', 'md5', 's3_status') {
   // Given a path to a file on disk (typically written to a temporary location for the
   // duration of the request), will do the work to generate a Blob instance with the
   // appropriate SHA and binary content information. Does _not_ save it to the database;

lib/model/frames/submission.js

@@ -102,7 +102,7 @@ Submission.Extended = Frame.define('formVersion', readable);
 
 Submission.Xml = Frame.define(table('submission_defs', 'xml'), 'xml');
 
-Submission.Encryption = Frame.define(into('encryption'), 'encHasData', 'encData', 'encIndex',  'encKeyId');
+Submission.Encryption = Frame.define(into('encryption'), 'encHasData', 'encData', 'encSha', 'encMd5', 'encIndex', 'encKeyId', 'encS3Status');
 
 Submission.Exports = Frame.define(into('exports'), 'formVersion');