feat: POST /eligible-deals-batch

api/index.js

@@ -36,6 +36,8 @@ const handler = async (req, res, client, domain) => {
     await getSummaryOfEligibleDealsForAllocator(req, res, client, segs[1])
   } else if (segs[0] === 'inspect-request' && req.method === 'GET') {
     await inspectRequest(req, res)
+  } else if (segs[0] === 'ingest-eligible-deals' && req.method === 'POST') {
+    await ingestEligibleDeals(req, res, client)
   } else {
     notFound(res)
   }
@@ -399,6 +401,51 @@ export const inspectRequest = async (req, res) => {
   })
 }
 
+export const ingestEligibleDeals = async (req, res, client) => {
+  const body = await getRawBody(req, { limit: '100mb' })
+  const deals = JSON.parse(body)
+  assert(Array.isArray(deals), 400, 'Invalid JSON Body, must be an array')
+  for (const d of deals) {
+    validate(d, 'clientId', { type: 'string', required: true })
+    validate(d, 'minerId', { type: 'string', required: true })
+    validate(d, 'pieceCid', { type: 'string', required: true })
+    validate(d, 'pieceSize', { type: 'string', required: true })
+    validate(d, 'payloadCid', { type: 'string', required: true })
+    validate(d, 'expiresAt', { type: 'date', required: true })
+  }
+
+  const { rowCount: ingested } = await client.query(`
+    INSERT INTO eligible_deals (
+      client_id,
+      miner_id,
+      piece_cid,
+      piece_size,
+      payload_cid,
+      expires_at,
+      sourced_from_f05_state
+    ) VALUES (
+      unnest($1::TEXT[]),
+      unnest($2::TEXT[]),
+      unnest($3::TEXT[]),
+      unnest($4::BIGINT[]),
+      unnest($5::TEXT[]),
+      unnest($6::DATE[]),
+      false
+    ) ON CONFLICT DO NOTHING`, [
+    deals.map(d => d.clientId),
+    deals.map(d => d.minerId),
+    deals.map(d => d.pieceCid),
+    deals.map(d => d.pieceSize),
+    deals.map(d => d.payloadCid),
+    deals.map(d => d.expiresAt)
+  ])
+
+  json(res, {
+    ingested,
+    skipped: deals.length - ingested
+  })
+}
+
 export const createHandler = async ({
   client,
   logger,

api/test/test.js

@@ -773,4 +773,88 @@ describe('Routes', () => {
       })
     })
   })
+
+  describe('POST /ingest-eligible-deals', () => {
+    beforeEach(async () => {
+      await client.query(
+        'DELETE FROM eligible_deals WHERE miner_id = $1',
+        ['f000']
+      )
+    })
+
+    it('ingests new deals', async () => {
+      const deals = [{
+        minerId: 'f000',
+        clientId: 'f001',
+        pieceCid: 'bagaone',
+        pieceSize: '34359738368',
+        payloadCid: 'bafyone',
+        expiresAt: '2100-01-01'
+      }]
+
+      const res = await fetch(`${spark}/ingest-eligible-deals`, {
+        method: 'POST',
+        body: JSON.stringify(deals)
+      })
+      await assertResponseStatus(res, 200)
+      const body = await res.json()
+
+      assert.deepStrictEqual(body, { ingested: 1, skipped: 0 })
+
+      const { rows } = await client.query(
+        'SELECT * FROM eligible_deals WHERE miner_id = $1',
+        ['f000']
+      )
+      assert.deepStrictEqual(rows, [{
+        miner_id: 'f000',
+        client_id: 'f001',
+        piece_cid: 'bagaone',
+        piece_size: '34359738368',
+        payload_cid: 'bafyone',
+        expires_at: new Date('2100-01-01'),
+        sourced_from_f05_state: false
+      }])
+    })
+
+    it('skips deals that were already ingested from f05 state', async () => {
+      const { rows: [f05Deal] } = await client.query(
+        'SELECT * FROM eligible_deals WHERE sourced_from_f05_state = TRUE LIMIT 1'
+      )
+
+      const res = await fetch(`${spark}/ingest-eligible-deals`, {
+        method: 'POST',
+        body: JSON.stringify([{
+          minerId: f05Deal.miner_id,
+          clientId: f05Deal.client_id,
+          pieceCid: f05Deal.piece_cid,
+          pieceSize: f05Deal.piece_size,
+          payloadCid: f05Deal.payload_cid,
+          expiresAt: f05Deal.expires_at.toISOString()
+        }])
+      })
+      await assertResponseStatus(res, 200)
+      const body = await res.json()
+
+      assert.deepStrictEqual(body, { ingested: 0, skipped: 1 })
+
+      const { rows } = await client.query(`
+        SELECT * FROM eligible_deals WHERE
+          miner_id = $1
+          AND client_id = $2
+          AND piece_cid = $3
+          AND piece_size = $4
+      `, [
+        f05Deal.miner_id,
+        f05Deal.client_id,
+        f05Deal.piece_cid,
+        f05Deal.piece_size
+      ])
+
+      assert.deepStrictEqual(rows, [f05Deal])
+    })
+
+    it.skip('rejects unauthorized requests', async () => {
+      // TODO
+    })
+  })
 })