rowStreamToOData(): reject unmatched repeatIds

lib/formats/odata.js

@@ -440,6 +440,8 @@ const rowStreamToOData = (fields, table, domain, originalUrl, query, inStream, t
     flush(done) {
       // flush is called just before the transform stream is done and closed; we write
       // our footer information, close the object, and tell the stream we are done.
+      if (!cursorPredicate) return this.destroy(Problem.user.odataRepeatIdNotFound());
+
       this.push((added === 0) ? '{"value":[],' : '],'); // open object or close row array.
 
       // if we were given an explicit count, use it from here out, to create

lib/http/endpoint.js

@@ -33,6 +33,15 @@ const Problem = require('../util/problem');
 // ENDPOINT COMMON UTILS
 // we put these first to appease eslint.
 
+const writeProblemJson = (response, error) => {
+  // we already have a publicly-consumable error object.
+  response.status(error.httpCode).type('application/json').send({
+    message: error.message,
+    code: error.problemCode,
+    details: error.problemDetails
+  });
+};
+
 // check if a string matches an expected mime type
 const isJsonType = (x) => /(^|,)(application\/json|json)($|;|,)/i.test(x);
 const isXmlType = (x) => /(^|,)(application\/(atom(svc)?\+)?xml|atom|xml)($|;|,)/i.test(x);
@@ -191,8 +200,12 @@ const endpointBase = ({ preprocessor = noop, before = noop, resultWriter, errorW
 ////////////////////////////////////////////////////////////////////////////////
 // ENDPOINT FORMAT SPECIALIZATIONS
 
-const streamErrorHandler = (response) => () => {
-  response.addTrailers({ Status: 'Error' }); // TODO: improve response content.
+const streamErrorHandler = (response) => err => {
+  if (err.isProblem && !response.headersSent && isJsonType(response.get('Content-Type'))) {
+    writeProblemJson(response, err);
+  } else {
+    response.addTrailers({ Status: 'Error' }); // TODO: improve response content.
+  }
 };
 const pipelineResult = (result, response, next) => {
   if (result instanceof PartialPipe) {
@@ -230,12 +243,7 @@ const defaultErrorWriter = (error, request, response) => {
   }
 
   if (error?.isProblem === true) {
-    // we already have a publicly-consumable error object.
-    response.status(error.httpCode).type('application/json').send({
-      message: error.message,
-      code: error.problemCode,
-      details: error.problemDetails
-    });
+    writeProblemJson(response, error);
   } else {
     debugger; // trip debugger if attached.
     process.stderr.write(inspect(error));

test/integration/api/odata.js

@@ -1888,6 +1888,27 @@ describe('api: /forms/:id.svc', () => {
         });
     }));
 
+    it('should reject unmatched repeatId', testService(async (service) => {
+      const asAlice = await withSubmissions(service, identity);
+
+      await asAlice.get('/v1/projects/1/forms/withrepeat.svc/Submissions.children.child?$top=2')
+        .expect(200)
+        .then(({ body }) => {
+          body.value[0].name.should.be.eql('Candace');
+          body.value[1].name.should.be.eql('Billy');
+          body['@odata.nextLink'].should.eql('http://localhost:8989/v1/projects/1/forms/withrepeat.svc/Submissions.children.child?%24top=2&%24skiptoken=01eyJyZXBlYXRJZCI6IjUyZWZmOWVhODI1NTAxODM4ODBiOWQ2NGMyMDQ4NzY0MmZhNmU2MGMifQ%3D%3D');
+          body['@odata.nextLink'].should.have.skiptoken({ repeatId: '52eff9ea82550183880b9d64c20487642fa6e60c' });
+          return body['@odata.nextLink'];
+        });
+
+      const skiptoken = '01' + encodeURIComponent(Buffer.from(JSON.stringify({ repeatId: 'nonsense' })).toString('base64'));
+      await asAlice.get(`/v1/projects/1/forms/withrepeat.svc/Submissions.children.child?%24top=2&%24skiptoken=${skiptoken}`)
+        .expect(400)
+        .then(({ body }) => {
+          body.should.deepEqual({ code: 400.34, message: 'Record associated with the provided $skiptoken not found.' });
+        });
+    }));
+
     it('should limit and filter subtable', testService(async (service) => {
       const asAlice = await withSubmissions(service, identity);
 

test/unit/formats/odata.js

@@ -852,6 +852,36 @@ describe('odata message composition', () => {
             done();
           })));
       });
+
+      [
+        { instanceId: 'two' },
+        { instanceId: 'two', repeatId: '' },
+        { instanceId: 'two', repeatId: 'this should probably be rejected' },
+        { instanceId: 'two', repeatId: '0000000000000000000000000000000000000000' },
+      ].forEach(skipToken => {
+        it(`should reject bad skipToken '${JSON.stringify(skipToken)}'`, done => {
+          const query = { $skiptoken: QueryOptions.getSkiptoken(skipToken) };
+          const inRows = streamTest.fromObjects([
+            mockSubmission('one', testData.instances.withrepeat.one),
+            mockSubmission('two', testData.instances.withrepeat.two),
+            mockSubmission('three', testData.instances.withrepeat.three)
+          ]);
+
+          fieldsFor(testData.forms.withrepeat)
+            .then((fields) => rowStreamToOData(fields, 'Submissions.children.child', 'http://localhost:8989', '/withrepeat.svc/Submissions.children.child?$skip=1&$top=1', query, inRows))
+            .then((stream) => {
+              stream.streams.at(-1).on('error', err => {
+                should(err).be.an.Error();
+                err.message.should.equal('Record associated with the provided $skiptoken not found.');
+                done();
+              });
+              return stream.pipe(streamTest.toText(err => {
+                if (err) return done(err);
+                done('pipe should not have completed');
+              }));
+            }, done);
+        });
+      });
     });
   });