feat: add repo removeById (multi)

.changeset/major-times-hammer.md

@@ -0,0 +1,5 @@
+---
+"@effect-app/infra": minor
+---
+
+feat: add repo removeById (multi)

packages/infra/src/Model/Repository/internal/internal.ts

@@ -57,15 +57,15 @@ export function makeRepoInternal<
           schemaContext?: Context.Context<RCtx>
           makeInitial?: Effect.Effect<readonly T[], E, RInitial> | undefined
           config?: Omit<StoreConfig<Encoded>, "partitionValue"> & {
-            partitionValue?: (a: Encoded) => string
+            partitionValue?: (e?: Encoded) => string
           }
         }
         : {
           schemaContext?: Context.Context<RCtx>
           publishEvents: (evt: NonEmptyReadonlyArray<Evt>) => Effect.Effect<void, never, RPublish>
           makeInitial?: Effect.Effect<readonly T[], E, RInitial> | undefined
           config?: Omit<StoreConfig<Encoded>, "partitionValue"> & {
-            partitionValue?: (a: Encoded) => string
+            partitionValue?: (e?: Encoded) => string
           }
         }
     ) {
@@ -125,6 +125,28 @@ export function makeRepoInternal<
                 : s.pipe(S.pick(idKey as any))
             })
           const encodeId = flow(S.encode(i), provideRctx)
+          const idOnly: S.Schema<T[IdKey], Encoded[IdKey], never> = ("fields" in fieldsSchema
+            ? S.Struct(fieldsSchema["fields"]) as unknown as typeof schema
+            : schema)
+            .pipe((_) => {
+              let ast = _.ast
+              if (ast._tag === "Declaration") ast = ast.typeParameters[0]!
+
+              const s = S.make(ast) as unknown as Schema<T, Encoded, R> & { fields: any }
+
+              return ast._tag === "Union"
+                // we need to get the TypeLiteral, incase of class it's behind a transform...
+                ? S.Union(
+                  ...ast.types.map((_) =>
+                    (S.make(_._tag === "Transformation" ? _.from : _) as unknown as Schema<T, Encoded> & {
+                      fields: any
+                    })
+                      .pipe((_) => _.fields[idKey])
+                  )
+                )
+                : s.fields[idKey]
+            })
+          const encodeIdOnly = flow(S.encode(idOnly), provideRctx)
           const findEId = Effect.fnUntraced(function*(id: Encoded[IdKey]) {
             yield* Effect.annotateCurrentSpan({ itemId: id })
 
@@ -210,6 +232,16 @@ export function makeRepoInternal<
             yield* changeFeed.publish([it, "remove"])
           })
 
+          const removeById = Effect.fn("removeById")(function*(...ids: NonEmptyReadonlyArray<T[IdKey]>) {
+            const { set } = yield* cms
+            const eids = yield* Effect.forEach(ids, (_) => encodeIdOnly(_)).pipe(Effect.orDie)
+            yield* store.batchRemove(eids)
+            for (const id of eids) {
+              set(id, undefined)
+            }
+            yield* changeFeed.publish([[], "remove"])
+          })
+
           const parseMany = (items: readonly PM[]) =>
             Effect
               .flatMap(cms, (cm) =>
@@ -311,6 +343,7 @@ export function makeRepoInternal<
             all,
             saveAndPublish,
             removeAndPublish,
+            removeById,
             queryRaw(schema, q) {
               const dec = S.decode(S.Array(schema))
               return store.queryRaw(q).pipe(Effect.flatMap(dec))
@@ -392,7 +425,7 @@ export function makeStore<Encoded extends FieldValues>() {
     function makeStore<RInitial = never, EInitial = never>(
       makeInitial?: Effect.Effect<readonly T[], EInitial, RInitial>,
       config?: Omit<StoreConfig<Encoded>, "partitionValue"> & {
-        partitionValue?: (a: Encoded) => string
+        partitionValue?: (e?: Encoded) => string
       }
     ) {
       function encodeToEncoded() {
@@ -454,29 +487,29 @@ export interface Repos<
     args: [Evt] extends [never] ? {
         makeInitial?: Effect.Effect<readonly T[], E, RInitial> | undefined
         config?: Omit<StoreConfig<Encoded>, "partitionValue"> & {
-          partitionValue?: (a: Encoded) => string
+          partitionValue?: (e?: Encoded) => string
         }
       }
       : {
         publishEvents: (evt: NonEmptyReadonlyArray<Evt>) => Effect.Effect<void, never, R2>
         makeInitial?: Effect.Effect<readonly T[], E, RInitial> | undefined
         config?: Omit<StoreConfig<Encoded>, "partitionValue"> & {
-          partitionValue?: (a: Encoded) => string
+          partitionValue?: (e?: Encoded) => string
         }
       }
   ): Effect.Effect<Repository<T, Encoded, Evt, ItemType, IdKey, RSchema, RPublish>, E, StoreMaker | RInitial | R2>
   makeWith<Out, RInitial = never, E = never, R2 = never>(
     args: [Evt] extends [never] ? {
         makeInitial?: Effect.Effect<readonly T[], E, RInitial> | undefined
         config?: Omit<StoreConfig<Encoded>, "partitionValue"> & {
-          partitionValue?: (a: Encoded) => string
+          partitionValue?: (e?: Encoded) => string
         }
       }
       : {
         publishEvents: (evt: NonEmptyReadonlyArray<Evt>) => Effect.Effect<void, never, R2>
         makeInitial?: Effect.Effect<readonly T[], E, RInitial> | undefined
         config?: Omit<StoreConfig<Encoded>, "partitionValue"> & {
-          partitionValue?: (a: Encoded) => string
+          partitionValue?: (e?: Encoded) => string
         }
       },
     f: (r: Repository<T, Encoded, Evt, ItemType, IdKey, RSchema, RPublish>) => Out

packages/infra/src/Model/Repository/makeRepo.ts

@@ -38,7 +38,7 @@ export interface RepositoryOptions<
    */
   jitM?: (pm: Encoded) => Encoded
   config?: Omit<StoreConfig<Encoded>, "partitionValue"> & {
-    partitionValue?: (a: Encoded) => string
+    partitionValue?: (e?: Encoded) => string
   }
   /**
    * Optional handler to be able to publish events after successfull save.

packages/infra/src/Model/Repository/service.ts

@@ -29,6 +29,8 @@ export interface Repository<
     events?: Iterable<Evt>
   ) => Effect.Effect<void, never, RSchema | RPublish>
 
+  readonly removeById: (...id: readonly T[IdKey][]) => Effect.Effect<void>
+
   readonly queryRaw: <T, Out, R>(
     schema: S.Schema<T, Out, R>,
     raw: RawQuery<Encoded, Out>

packages/infra/src/Model/query.ts

@@ -2,7 +2,7 @@ export * from "./query/dsl.js"
 export * from "./query/new-kid-interpreter.js"
 
 export interface RawQuery<Encoded, Out> {
-  cosmos: (vals: { importedMarkerId: string; name: string }) => {
+  cosmos: (vals: { name: string }) => {
     query: string
     parameters: {
       name: string

packages/infra/src/Store/Cosmos.ts

@@ -2,7 +2,7 @@
 
 import { Array, Chunk, Duration, Effect, Layer, type NonEmptyReadonlyArray, Option, pipe, Redacted, Struct } from "effect-app"
 import { toNonEmptyArray } from "effect-app/Array"
-import { dropUndefinedT } from "effect-app/utils"
+import { dropUndefinedT, mutable } from "effect-app/utils"
 import { CosmosClient, CosmosClientLayer } from "../adapters/cosmos-client.js"
 import { OptimisticConcurrencyException } from "../errors.js"
 import { InfraLogger } from "../logger.js"
@@ -50,6 +50,8 @@ function makeCosmosStore({ prefix }: StorageConfig) {
             }))
           )
 
+          const mainPartitionKey = config?.partitionValue() ?? "primary"
+
           const defaultValues = config?.defaultValues ?? {}
           const container = db.container(containerId)
           const bulk = container.items.bulk.bind(container.items)
@@ -228,14 +230,14 @@ function makeCosmosStore({ prefix }: StorageConfig) {
           const s: Store<IdKey, Encoded> = {
             queryRaw: <Out>(query: RawQuery<Encoded, Out>) =>
               Effect
-                .sync(() => query.cosmos({ importedMarkerId, name }))
+                .sync(() => query.cosmos({ name }))
                 .pipe(
                   Effect.tap((q) => logQuery(q)),
                   Effect.flatMap((q) =>
                     Effect.promise(() =>
                       container
                         .items
-                        .query<Out>(q, { partitionKey: "primary" })
+                        .query<Out>(q, { partitionKey: mainPartitionKey })
                         .fetchAll()
                         .then(({ resources }) =>
                           resources.map(
@@ -250,18 +252,28 @@ function makeCosmosStore({ prefix }: StorageConfig) {
                       attributes: { "repository.container_id": containerId, "repository.model_name": name }
                     })
                 ),
+            batchRemove: (ids) =>
+              Effect.promise(() =>
+                execBatch(mutable(ids.map((id) =>
+                  dropUndefinedT({
+                    operationType: "Delete" as const,
+                    id,
+                    partitionKey: config?.partitionValue({ [idKey]: id } as Encoded)
+                  })
+                )))
+              ),
             all: Effect
               .sync(() => ({
-                query: `SELECT * FROM ${name} f WHERE f.id != @id`,
-                parameters: [{ name: "@id", value: importedMarkerId }]
+                query: `SELECT * FROM ${name}`,
+                parameters: []
               }))
               .pipe(
                 Effect.tap((q) => logQuery(q)),
                 Effect.flatMap((q) =>
                   Effect.promise(() =>
                     container
                       .items
-                      .query<PMCosmos>(q)
+                      .query<PMCosmos>(q, { partitionKey: mainPartitionKey })
                       .fetchAll()
                       .then(({ resources }) =>
                         resources.map(
@@ -308,7 +320,7 @@ function makeCosmosStore({ prefix }: StorageConfig) {
                         f.select
                           ? container
                             .items
-                            .query<M>(q)
+                            .query<M>(q, { partitionKey: mainPartitionKey })
                             .fetchAll()
                             .then(({ resources }) =>
                               resources.map((_) =>
@@ -323,7 +335,7 @@ function makeCosmosStore({ prefix }: StorageConfig) {
                             )
                           : container
                             .items
-                            .query<{ f: M }>(q)
+                            .query<{ f: M }>(q, { partitionKey: mainPartitionKey })
                             .fetchAll()
                             .then(({ resources }) =>
                               resources.map(({ f }) => ({ ...defaultValues, ...mapReverseId(f as any) }) as any)

packages/infra/src/Store/Cosmos/query.ts

@@ -39,7 +39,7 @@
   idKey: PropertyKey,
   filter: readonly FilterResult[],
   name: string,
   importedMarkerId: string,
   defaultValues: Record<string, unknown>,
   select?: NonEmptyReadonlyArray<string | { key: string; subKeys: readonly string[] }>,
   order?: NonEmptyReadonlyArray<{ key: string; direction: "ASC" | "DESC" }>,
@@ -294,11 +294,10 @@
     }
     FROM ${name} f
 
-    WHERE f.id != @id ${filter.length ? `AND (${print(filter, values.map((_) => _.value), null, false)})` : ""}
+    ${filter.length ? `WHERE (${print(filter, values.map((_) => _.value), null, false)})` : ""}
     ${order ? `ORDER BY ${order.map((_) => `${dottedToAccess(`f.${_.key}`)} ${_.direction}`).join(", ")}` : ""}
     ${skip !== undefined || limit !== undefined ? `OFFSET ${skip ?? 0} LIMIT ${limit ?? 999999}` : ""}`,
     parameters: [
-      { name: "@id", value: importedMarkerId },
       ...values
         .flatMap((x, i) =>
           [{

packages/infra/src/Store/Disk.ts

@@ -97,6 +97,10 @@ function makeDiskStoreInt<IdKey extends keyof Encoded, Encoded extends FieldValu
         store.batchSet,
         Effect.tap(flushToDiskInBackground)
       ),
+      batchRemove: flow(
+        store.batchRemove,
+        Effect.tap(flushToDiskInBackground)
+      ),
       bulkSet: flow(
         store.bulkSet,
         Effect.tap(flushToDiskInBackground)
@@ -173,6 +177,7 @@ export function makeDiskStore({ prefix }: StorageConfig, dir: string) {
             set: (...args) => Effect.flatMap(getStore, (_) => _.set(...args)),
             batchSet: (...args) => Effect.flatMap(getStore, (_) => _.batchSet(...args)),
             bulkSet: (...args) => Effect.flatMap(getStore, (_) => _.bulkSet(...args)),
+            batchRemove: (...args) => Effect.flatMap(getStore, (_) => _.batchRemove(...args)),
             remove: (...args) => Effect.flatMap(getStore, (_) => _.remove(...args)),
             queryRaw: (...args) => Effect.flatMap(getStore, (_) => _.queryRaw(...args))
           }

packages/infra/src/Store/Memory.ts

@@ -137,6 +137,23 @@ export function makeMemoryStoreInt<IdKey extends keyof Encoded, Encoded extends
             .map((_) => _),
           withPermit
         )
+
+    const batchRemove = (items: NonEmptyReadonlyArray<Encoded[IdKey]>) =>
+      Ref
+        .get(store)
+        .pipe(
+          Effect
+            .map((m) => {
+              const mut = m as Map<string, PM>
+              items.forEach((e) => mut.delete(e[idKey]))
+              return mut
+            }),
+          Effect
+            .flatMap((_) => Ref.set(store, _))
+        )
+        .pipe(
+          withPermit
+        )
     const s: Store<IdKey, Encoded> = {
       queryRaw: (query) =>
         all
@@ -199,6 +216,21 @@ export function makeMemoryStoreInt<IdKey extends keyof Encoded, Encoded extends
                 attributes: { "repository.model_name": modelName, "repository.namespace": namespace }
               })
           ),
+      batchRemove: (items: NonEmptyReadonlyArray<Encoded[IdKey]>) =>
+        pipe(
+          Effect
+            .sync(() => items)
+            // align with CosmosDB
+            .pipe(
+              Effect.filterOrDieMessage((_) => _.length <= 100, "BatchRemove: a batch may not exceed 100 items"),
+              Effect.andThen(batchRemove),
+              Effect
+                .withSpan("Memory.batchRemove [effect-app/infra/Store]", {
+                  captureStackTrace: false,
+                  attributes: { "repository.model_name": modelName, "repository.namespace": namespace }
+                })
+            )
+        ),
       batchSet: (items: readonly [PM, ...PM[]]) =>
         pipe(
           Effect
@@ -286,6 +318,7 @@ export const makeMemoryStore = () => ({
         set: (...args) => Effect.flatMap(getStore, (_) => _.set(...args)),
         batchSet: (...args) => Effect.flatMap(getStore, (_) => _.batchSet(...args)),
         bulkSet: (...args) => Effect.flatMap(getStore, (_) => _.bulkSet(...args)),
+        batchRemove: (...args) => Effect.flatMap(getStore, (_) => _.batchRemove(...args)),
         remove: (...args) => Effect.flatMap(getStore, (_) => _.remove(...args))
       }
       return s

packages/infra/src/Store/service.ts

@@ -8,7 +8,7 @@ import type { FieldPath } from "../Model/filter/types/path/index.js"
 import { type RawQuery } from "../Model/query.js"
 
 export interface StoreConfig<E> {
-  partitionValue: (e: E) => string | undefined
+  partitionValue: (e?: E) => string
   /**
    * Primarily used for testing, creating namespaces in the database to separate data e.g to run multiple tests in isolation within the same database
    * currently only supported in disk/memory. CosmosDB is TODO.
@@ -89,6 +89,9 @@ export interface Store<
    * Requires the Encoded type, not Id, because various stores may need to calculate e.g partition keys.
    */
   remove: (e: Encoded) => Effect.Effect<void>
+  batchRemove: (ids: NonEmptyReadonlyArray<Encoded[IdKey]>) => Effect.Effect<void>
+  // TODO: only accept where filter, nothing else
+  // filterRemove: FilterFunc<Encoded>
 
   queryRaw: <Out>(query: RawQuery<Encoded, Out>) => Effect.Effect<readonly Out[]>
 }