diff --git a/.changeset/major-times-hammer.md b/.changeset/major-times-hammer.md new file mode 100644 index 0000000000..6f3b271ca6 --- /dev/null +++ b/.changeset/major-times-hammer.md @@ -0,0 +1,5 @@ +--- +"@effect-app/infra": minor +--- + +feat: add repo removeById (multi) diff --git a/packages/infra/src/Model/Repository/internal/internal.ts b/packages/infra/src/Model/Repository/internal/internal.ts index 3e925917da..d9447fc27e 100644 --- a/packages/infra/src/Model/Repository/internal/internal.ts +++ b/packages/infra/src/Model/Repository/internal/internal.ts @@ -57,7 +57,7 @@ export function makeRepoInternal< schemaContext?: Context.Context makeInitial?: Effect.Effect | undefined config?: Omit, "partitionValue"> & { - partitionValue?: (a: Encoded) => string + partitionValue?: (e?: Encoded) => string } } : { @@ -65,7 +65,7 @@ export function makeRepoInternal< publishEvents: (evt: NonEmptyReadonlyArray) => Effect.Effect makeInitial?: Effect.Effect | undefined config?: Omit, "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 = ("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 & { 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 & { + 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) { + 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() { function makeStore( makeInitial?: Effect.Effect, config?: Omit, "partitionValue"> & { - partitionValue?: (a: Encoded) => string + partitionValue?: (e?: Encoded) => string } ) { function encodeToEncoded() { @@ -454,14 +487,14 @@ export interface Repos< args: [Evt] extends [never] ? { makeInitial?: Effect.Effect | undefined config?: Omit, "partitionValue"> & { - partitionValue?: (a: Encoded) => string + partitionValue?: (e?: Encoded) => string } } : { publishEvents: (evt: NonEmptyReadonlyArray) => Effect.Effect makeInitial?: Effect.Effect | undefined config?: Omit, "partitionValue"> & { - partitionValue?: (a: Encoded) => string + partitionValue?: (e?: Encoded) => string } } ): Effect.Effect, E, StoreMaker | RInitial | R2> @@ -469,14 +502,14 @@ export interface Repos< args: [Evt] extends [never] ? { makeInitial?: Effect.Effect | undefined config?: Omit, "partitionValue"> & { - partitionValue?: (a: Encoded) => string + partitionValue?: (e?: Encoded) => string } } : { publishEvents: (evt: NonEmptyReadonlyArray) => Effect.Effect makeInitial?: Effect.Effect | undefined config?: Omit, "partitionValue"> & { - partitionValue?: (a: Encoded) => string + partitionValue?: (e?: Encoded) => string } }, f: (r: Repository) => Out diff --git a/packages/infra/src/Model/Repository/makeRepo.ts b/packages/infra/src/Model/Repository/makeRepo.ts index 08cc197560..46bfa0c040 100644 --- a/packages/infra/src/Model/Repository/makeRepo.ts +++ b/packages/infra/src/Model/Repository/makeRepo.ts @@ -38,7 +38,7 @@ export interface RepositoryOptions< */ jitM?: (pm: Encoded) => Encoded config?: Omit, "partitionValue"> & { - partitionValue?: (a: Encoded) => string + partitionValue?: (e?: Encoded) => string } /** * Optional handler to be able to publish events after successfull save. diff --git a/packages/infra/src/Model/Repository/service.ts b/packages/infra/src/Model/Repository/service.ts index 2890400b93..02f1f1ce96 100644 --- a/packages/infra/src/Model/Repository/service.ts +++ b/packages/infra/src/Model/Repository/service.ts @@ -29,6 +29,8 @@ export interface Repository< events?: Iterable ) => Effect.Effect + readonly removeById: (...id: readonly T[IdKey][]) => Effect.Effect + readonly queryRaw: ( schema: S.Schema, raw: RawQuery diff --git a/packages/infra/src/Model/query.ts b/packages/infra/src/Model/query.ts index 327113ce3f..4d8d51b65e 100644 --- a/packages/infra/src/Model/query.ts +++ b/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 { - cosmos: (vals: { importedMarkerId: string; name: string }) => { + cosmos: (vals: { name: string }) => { query: string parameters: { name: string diff --git a/packages/infra/src/Store/Cosmos.ts b/packages/infra/src/Store/Cosmos.ts index 7230f22ad3..4a74c82639 100644 --- a/packages/infra/src/Store/Cosmos.ts +++ b/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 = { queryRaw: (query: RawQuery) => Effect - .sync(() => query.cosmos({ importedMarkerId, name })) + .sync(() => query.cosmos({ name })) .pipe( Effect.tap((q) => logQuery(q)), Effect.flatMap((q) => Effect.promise(() => container .items - .query(q, { partitionKey: "primary" }) + .query(q, { partitionKey: mainPartitionKey }) .fetchAll() .then(({ resources }) => resources.map( @@ -250,10 +252,20 @@ 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)), @@ -261,7 +273,7 @@ function makeCosmosStore({ prefix }: StorageConfig) { Effect.promise(() => container .items - .query(q) + .query(q, { partitionKey: mainPartitionKey }) .fetchAll() .then(({ resources }) => resources.map( @@ -308,7 +320,7 @@ function makeCosmosStore({ prefix }: StorageConfig) { f.select ? container .items - .query(q) + .query(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) diff --git a/packages/infra/src/Store/Cosmos/query.ts b/packages/infra/src/Store/Cosmos/query.ts index ff8c1f807c..b4032991dd 100644 --- a/packages/infra/src/Store/Cosmos/query.ts +++ b/packages/infra/src/Store/Cosmos/query.ts @@ -294,11 +294,10 @@ export function buildWhereCosmosQuery3( } 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) => [{ diff --git a/packages/infra/src/Store/Disk.ts b/packages/infra/src/Store/Disk.ts index fb29037f70..258c35df29 100644 --- a/packages/infra/src/Store/Disk.ts +++ b/packages/infra/src/Store/Disk.ts @@ -97,6 +97,10 @@ function makeDiskStoreInt 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)) } diff --git a/packages/infra/src/Store/Memory.ts b/packages/infra/src/Store/Memory.ts index 9da46b8f68..03ebb3278e 100644 --- a/packages/infra/src/Store/Memory.ts +++ b/packages/infra/src/Store/Memory.ts @@ -137,6 +137,23 @@ export function makeMemoryStoreInt _), withPermit ) + + const batchRemove = (items: NonEmptyReadonlyArray) => + Ref + .get(store) + .pipe( + Effect + .map((m) => { + const mut = m as Map + items.forEach((e) => mut.delete(e[idKey])) + return mut + }), + Effect + .flatMap((_) => Ref.set(store, _)) + ) + .pipe( + withPermit + ) const s: Store = { queryRaw: (query) => all @@ -199,6 +216,21 @@ export function makeMemoryStoreInt) => + 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 diff --git a/packages/infra/src/Store/service.ts b/packages/infra/src/Store/service.ts index 65605a6404..81107e9afd 100644 --- a/packages/infra/src/Store/service.ts +++ b/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 { - 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 + batchRemove: (ids: NonEmptyReadonlyArray) => Effect.Effect + // TODO: only accept where filter, nothing else + // filterRemove: FilterFunc queryRaw: (query: RawQuery) => Effect.Effect }