refactor: Simplify bucket handling

rdf-connect · Oct 14, 2024 · e5d6bb0 · e5d6bb0
1 parent c0abf12
commit e5d6bb0
Show file tree

Hide file tree

Showing 7 changed files with 17 additions and 43 deletions.
diff --git a/src/extractor.ts b/src/extractor.ts
@@ -16,7 +16,7 @@ export type Record = {
 
 export type Bucket = {
     id: string;
-    stream: string;
+    streamId: string;
     immutable?: boolean;
     root?: boolean;
     empty?: boolean;

diff --git a/src/index.ts b/src/index.ts
@@ -114,26 +114,7 @@ async function handleBuckets(
     const buckets = extract.getBuckets();
 
     for (const bucket of buckets) {
-        const set: {
-            immutable?: boolean;
-            root?: boolean;
-            empty?: boolean;
-            members?: string[];
-        } = {};
-        if (bucket.root !== undefined) {
-            set.root = bucket.root;
-        }
-
-        if (bucket.empty !== undefined) {
-            set.empty = bucket.empty;
-            set.members = [];
-        }
-
-        if (bucket.immutable !== undefined) {
-            set.immutable = bucket.immutable;
-        }
-
-        await repository.handleBucket(bucket, set, operations);
+        await repository.handleBucket(bucket, operations);
     }
 }
 

diff --git a/src/repositories/MongoDBRepository.ts b/src/repositories/MongoDBRepository.ts
@@ -148,17 +148,20 @@ export class MongoDBRepository implements Repository {
 
     async handleBucket(
         bucket: Bucket,
-        parameters: any,
         bulk: AnyBulkWriteOperation<TREEFragment>[],
     ): Promise<void> {
+        if (bucket.empty) {
+            (bucket as TREEFragment).members = [];
+            delete bucket.empty;
+        }
         bulk.push({
             updateOne: {
                 filter: {
-                    streamId: bucket.stream,
+                    streamId: bucket.streamId,
                     id: bucket.id,
                 },
                 update: {
-                    $set: parameters,
+                    $set: bucket,
                 },
                 upsert: true,
             },

diff --git a/src/repositories/RedisRepository.ts b/src/repositories/RedisRepository.ts
@@ -120,31 +120,22 @@ export class RedisRepository implements Repository {
 
     async handleBucket(
         bucket: Bucket,
-        parameters: any,
         bulk: Promise<string | number | null>[],
     ): Promise<void> {
-        // If parameters contains `members`, use the handleMember method and remove the key.
-        if (parameters.members || parameters.empty) {
+        // If bucket contains `empty`, remove the members set
+        if (bucket.empty) {
             bulk.push(
                 this.client.del(
-                    `${this.index}:${encodeURIComponent(bucket.stream)}:${encodeURIComponent(bucket.id)}:members`,
+                    `${this.index}:${encodeURIComponent(bucket.streamId)}:${encodeURIComponent(bucket.id)}:members`,
                 ),
             );
-            for (const member of parameters.members) {
-                await this.handleMember(member, bucket.id, bulk);
-            }
-            delete parameters.members;
         }
 
-        // Set streamId and id parameters
-        parameters.streamId = bucket.stream;
-        parameters.id = bucket.id;
-
         bulk.push(
             this.client.json.set(
-                `${this.index}:${encodeURIComponent(bucket.stream)}:${encodeURIComponent(bucket.id)}`,
+                `${this.index}:${encodeURIComponent(bucket.streamId)}:${encodeURIComponent(bucket.id)}`,
                 ".",
-                parameters,
+                bucket,
             ),
         );
     }

diff --git a/src/repositories/Repository.ts b/src/repositories/Repository.ts
@@ -48,7 +48,6 @@ export interface Repository {
 
     handleBucket(
         bucket: Bucket,
-        parameters: any,
         bulk: IndexBulkOperations,
     ): Promise<void>;
 
@@ -64,14 +63,14 @@ export function getRepository(dbConfig: DBConfig): Repository {
     const url =
         dbConfig.url || env.DB_CONN_STRING || "mongodb://localhost:27017/ldes";
 
-    if (url.startsWith("mongodb://")) {
+    if (url.startsWith("mongodb")) {
         return new MongoDBRepository(
             url,
             dbConfig.metadata,
             dbConfig.data,
             dbConfig.index,
         );
-    } else if (url.startsWith("redis://")) {
+    } else if (url.startsWith("redis")) {
         return new RedisRepository(
             url,
             dbConfig.metadata,

diff --git a/src/shape.ttl b/src/shape.ttl
@@ -41,7 +41,7 @@
     sh:minCount 1;
     sh:maxCount 1;
   ], [
-    sh:name "stream";
+    sh:name "streamId";
     sh:path sds:stream;
     sh:datatype xsd:string;
     sh:minCount 1;

diff --git a/test/extractor.test.ts b/test/extractor.test.ts
@@ -107,7 +107,7 @@ _:n3-26 <http://www.w3.org/1999/02/22-rdf-syntax-ns#first> <http://qudt.org/1.1/
 
         const buckets = extract.getBuckets();
         expect(buckets.length).toBe(4);
-        expect(buckets.map((x) => x.stream)).toEqual([
+        expect(buckets.map((x) => x.streamId)).toEqual([
             "https://example.org/ns#",
             "https://example.org/ns#",
             "https://example.org/ns#",