Add implementation. Closes #1

.github/workflows/ci.yml

@@ -0,0 +1,11 @@
+name: CI
+on: [push]
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v3
+      - run: curl -L https://github.com/filecoin-station/zinnia/releases/download/v0.10.0/zinnia-linux-x64.tar.gz | tar -xz
+      - uses: actions/setup-node@v3
+      - run: npx standard
+      - run: ./zinnia run test.js

README.md

@@ -3,3 +3,16 @@ SP Retrieval Checker Module
 
 - [Roadmap](https://pl-strflt.notion.site/SPARK-Roadmap-ac729c11c49b409fbec54751d1bc6c8a)
 - [API](https://github.com/filecoin-station/spark-api)
+
+## Development
+
+Install [Zinnia CLI](https://github.com/filecoin-station/zinnia).
+
+```bash
+$ # Lint
+$ npx standard
+$ # Run module
+$ zinnia run main.js
+$ # Test module
+$ zinnia run test.js
+```

lib/spark.js

@@ -0,0 +1,134 @@
+/* global Zinnia */
+
+import { assertEquals } from 'zinnia:assert'
+
+const sleep = dt => new Promise(resolve => setTimeout(resolve, dt))
+
+// Create activity events when we bacome operational or fail, but only once
+export class ActivityState {
+  #ok = true
+
+  onError () {
+    if (this.#ok) {
+      this.#ok = false
+      Zinnia.activity.error('SPARK failed reporting retrieval')
+    }
+  }
+
+  onSuccess () {
+    if (!this.#ok) {
+      this.#ok = true
+      Zinnia.activity.success('SPARK retrieval reporting resumed')
+    }
+  }
+}
+
+export default class Spark {
+  #fetch
+  #activity = new ActivityState()
+
+  constructor ({ fetch = globalThis.fetch } = {}) {
+    this.#fetch = fetch
+  }
+
+  async getRetrieval () {
+    console.log('Geting retrieval...')
+    const res = await this.#fetch('https://spark.fly.dev/retrievals', {
+      method: 'POST'
+    })
+    const retrieval = await res.json()
+    console.log({ retrieval })
+    return retrieval
+  }
+
+  async fetchCAR (url) {
+    // TODO: Ensure these fields are persisted by `spark-api`
+    const stats = {
+      start: new Date(),
+      firstByte: null,
+      end: null,
+      byteLength: 0,
+      status: null
+    }
+    console.log('Fetching CAR...')
+
+    // Abort if no progress was made for 10 seconds
+    const controller = new AbortController()
+    const { signal } = controller
+    let timeout
+    const startTimeout = () => {
+      if (timeout) {
+        clearTimeout(timeout)
+      }
+      timeout = setTimeout(() => controller.abort(), 10_000)
+    }
+
+    try {
+      startTimeout()
+      const res = await this.#fetch(url, { signal })
+      stats.status = res.status
+
+      if (res.ok) {
+        startTimeout()
+        for await (const value of res.body) {
+          if (stats.firstByte === null) {
+            stats.firstByte = new Date()
+          }
+          stats.byteLength += value.byteLength
+          startTimeout()
+        }
+      }
+    } finally {
+      clearTimeout(timeout)
+    }
+
+    stats.end = new Date()
+    console.log(stats)
+    return stats
+  }
+
+  async submitRetrieval (id, stats) {
+    console.log('Submitting retrieval...')
+    const res = await this.#fetch(`https://spark.fly.dev/retrievals/${id}`, {
+      method: 'PATCH',
+      body: JSON.stringify(stats),
+      headers: {
+        'Content-Type': 'application/json'
+      }
+    })
+    assertEquals(res.status, 200)
+    console.log('Retrieval submitted')
+  }
+
+  async nextRetrieval () {
+    const retrieval = await this.getRetrieval()
+
+    let success = false
+    let stats
+    const url = `https://strn.pl/ipfs/${retrieval.cid}`
+    try {
+      stats = await this.fetchCAR(url)
+      success = true
+      this.#activity.onSuccess()
+    } catch (err) {
+      console.error(`Failed to fetch ${url}`)
+      console.error(err)
+      this.#activity.onError()
+    }
+
+    await this.submitRetrieval(retrieval.id, { success, ...stats })
+    Zinnia.jobCompleted()
+  }
+
+  async run () {
+    while (true) {
+      try {
+        await this.nextRetrieval()
+      } catch (err) {
+        Zinnia.activity.error('SPARK failed reporting retrieval')
+        console.error(err)
+      }
+      await sleep(1_000)
+    }
+  }
+}

main.js

@@ -0,0 +1,4 @@
+import Spark from './lib/spark.js'
+
+const spark = new Spark()
+await spark.run()

test.js

@@ -0,0 +1,2 @@
+import './test/integration.js'
+import './test/spark.js'

test/integration.js

@@ -0,0 +1,15 @@
+import Spark from '../lib/spark.js'
+import { test } from 'zinnia:test'
+
+test('integration', async () => {
+  const spark = new Spark()
+  for (let i = 0; i < 100; i++) {
+    try {
+      await spark.nextRetrieval()
+      return
+    } catch (err) {
+      console.error(err)
+    }
+  }
+  throw new Error('No retrieval succeeded')
+})

test/spark.js

@@ -0,0 +1,70 @@
+import Spark from '../lib/spark.js'
+import { test } from 'zinnia:test'
+import { assertInstanceOf, assertEquals } from 'zinnia:assert'
+
+test('getRetrieval', async () => {
+  const retrieval = { retrieval: 'retrieval' }
+  const requests = []
+  const fetch = async (url, opts) => {
+    requests.push({ url, opts })
+    return {
+      async json () {
+        return retrieval
+      }
+    }
+  }
+  const spark = new Spark({ fetch })
+  assertEquals(await spark.getRetrieval(), retrieval)
+  assertEquals(requests, [{
+    url: 'https://spark.fly.dev/retrievals',
+    opts: { method: 'POST' }
+  }])
+})
+
+// TODO: test more cases
+test('fetchCAR', async () => {
+  const URL = 'url'
+  const requests = []
+  const fetch = async url => {
+    requests.push({ url })
+    return {
+      status: 200,
+      ok: true,
+      body: (async function * () {
+        yield new Uint8Array([1, 2, 3])
+      })()
+    }
+  }
+  const spark = new Spark({ fetch })
+  const stats = await spark.fetchCAR(URL)
+  assertInstanceOf(stats.start, Date)
+  assertInstanceOf(stats.firstByte, Date)
+  assertInstanceOf(stats.end, Date)
+  assertEquals(stats.byteLength, 3)
+  assertEquals(stats.status, 200)
+  assertEquals(requests, [{ url: URL }])
+})
+
+test('submitRetrieval', async () => {
+  const requests = []
+  const fetch = async (url, opts) => {
+    requests.push({ url, opts })
+    assertEquals(url, 'https://spark.fly.dev/retrievals/0')
+    assertEquals(opts.method, 'PATCH')
+    assertEquals(JSON.parse(opts.body), { success: true })
+    assertEquals(opts.headers['Content-Type'], 'application/json')
+    return { status: 200 }
+  }
+  const spark = new Spark({ fetch })
+  await spark.submitRetrieval(0, { success: true })
+  assertEquals(requests, [
+    {
+      url: 'https://spark.fly.dev/retrievals/0',
+      opts: {
+        method: 'PATCH',
+        body: JSON.stringify({ success: true }),
+        headers: { 'Content-Type': 'application/json' }
+      }
+    }
+  ])
+})