Load ORCID iDs from Redis

package.json

@@ -22,7 +22,9 @@
     "typescript": "^5.3.2"
   },
   "dependencies": {
+    "@effect/schema": "^0.50.0",
     "effect": "^2.0.0-next.56",
+    "ioredis": "^5.3.2",
     "orcid-id-ts": "^0.1.2"
   }
 }

src/OrcidId.ts

@@ -1,6 +1,9 @@
+import { Schema } from '@effect/schema'
 import { Brand } from 'effect'
 import { isOrcid } from 'orcid-id-ts'
 
 export type OrcidId = Brand.Branded<string, 'OrcidId'>
 
 export const OrcidId = Brand.refined<OrcidId>(isOrcid, s => Brand.error(`Expected ${s} to be an ORCID iD`))
+
+export const OrcidIdSchema = Schema.string.pipe(Schema.fromBrand(OrcidId))

src/Program.ts

@@ -5,4 +5,4 @@ import * as Users from './Users.js'
 const processUser = (orcidId: OrcidId.OrcidId) =>
   Effect.logInfo('Processing user').pipe(Effect.annotateLogs('orcidId', orcidId))
 
-export const program = Users.getUsers.pipe(Stream.runForEach(processUser))
+export const program = Users.getUsers.pipe(Stream.runForEach(processUser)).pipe(Effect.scoped)

src/Redis.ts

@@ -0,0 +1,22 @@
+import { Schema } from '@effect/schema'
+import { Context, Data, Effect, type ReadonlyArray, Stream } from 'effect'
+import type IoRedis from 'ioredis'
+
+export type Redis = IoRedis.Redis
+
+export const Redis = Context.Tag<Redis>('IoRedis/Redis')
+
+export class RedisError extends Data.TaggedError('RedisError')<{
+  readonly error: unknown
+}> {}
+
+export const scanStream = (
+  options: Parameters<Redis['scanStream']>[0],
+): Stream.Stream<Redis, RedisError, ReadonlyArray<string>> =>
+  Stream.unwrap(
+    Effect.gen(function* (_) {
+      const redis = yield* _(Redis)
+
+      return Stream.fromAsyncIterable(redis.scanStream(options), error => new RedisError({ error }))
+    }),
+  ).pipe(Stream.map(Schema.decodeSync(Schema.array(Schema.string))))

src/Users.ts

@@ -1,4 +1,17 @@
-import { Stream } from 'effect'
+import { type ParseResult, Schema } from '@effect/schema'
+import { ReadonlyArray, type Scope, Stream, String, identity } from 'effect'
 import * as OrcidId from './OrcidId.js'
+import * as Redis from './Redis.js'
 
-export const getUsers = Stream.fromIterable([OrcidId.OrcidId('0000-0001-8778-8651')])
+export const getUsers: Stream.Stream<
+  Redis.Redis | Scope.Scope,
+  Redis.RedisError | ParseResult.ParseError,
+  OrcidId.OrcidId
+> = Redis.scanStream({
+  match: 'orcid-token:*',
+}).pipe(
+  Stream.mapConcat(identity),
+  Stream.map(String.split(':')),
+  Stream.map(ReadonlyArray.lastNonEmpty),
+  Stream.flatMap(Schema.decodeEither(OrcidId.OrcidIdSchema)),
+)

src/index.ts

@@ -1,4 +1,19 @@
-import { Effect } from 'effect'
+import { Effect, Layer } from 'effect'
+import IoRedis from 'ioredis'
 import { program } from './Program.js'
+import * as Redis from './Redis.js'
 
-await Effect.runPromise(program)
+const RedisLive = Layer.effect(
+  Redis.Redis,
+  Effect.gen(function* (_) {
+    const redis = new IoRedis.Redis()
+
+    yield* _(Effect.addFinalizer(() => Effect.sync(() => redis.disconnect())))
+
+    return redis
+  }),
+)
+
+const runnable = Effect.provide(program, RedisLive).pipe(Effect.scoped)
+
+await Effect.runPromise(runnable)