feat(lambda): functionality for updating ID, values, and deleting packages

lib/lambda/search.test.ts

@@ -1,4 +1,4 @@
-import { describe, it, expect, vi, beforeEach } from "vitest";
+import { describe, it, expect, vi, beforeEach, Mock } from "vitest";
 import { APIGatewayEvent } from "aws-lambda";
 import { handler } from "./search";
 import { response } from "libs/handler-lib";
@@ -52,8 +52,8 @@ describe("getSearchData Handler", () => {
       },
     };
 
-    (os.search as vi.Mock).mockResolvedValueOnce(mockResults);
-    (getAppkChildren as vi.Mock).mockResolvedValueOnce({
+    (os.search as Mock).mockResolvedValueOnce(mockResults);
+    (getAppkChildren as Mock).mockResolvedValueOnce({
       hits: {
         hits: [{ _id: "2", _source: { child: "child-data" } }],
       },
@@ -62,7 +62,7 @@ describe("getSearchData Handler", () => {
     const event = {
       pathParameters: { index: "main" },
       body: JSON.stringify({ query: { match_all: {} } }),
-    } as APIGatewayEvent;
+    } as unknown as APIGatewayEvent;
 
     await handler(event);
 

lib/lambda/search.ts

@@ -23,6 +23,8 @@ export const getSearchData = async (event: APIGatewayEvent) => {
     query.query = query?.query || {};
     query.query.bool = query.query?.bool || {};
     query.query.bool.must = query.query.bool?.must || [];
+    query.query.bool.must_not = query.query.bool?.must_not || [];
+    query.query.bool.must_not.push({ term: { deleted: true } });
 
     const stateFilter = await getStateFilter(event);
     if (stateFilter) {

lib/lambda/sinkChangelog.ts

@@ -8,6 +8,12 @@ import {
   logError,
 } from "../libs/sink-lib";
 import { Index } from "shared-types/opensearch";
+import {
+  transformDeleteSchema,
+  transformUpdateValuesSchema,
+  transformedUpdateIdSchema,
+} from "./update/adminChangeSchemas";
+import { getPackageChangelog } from "lib/libs/api/package";
 const osDomain = process.env.osDomain;
 if (!osDomain) {
   throw new Error("Missing required environment variable(s)");
@@ -61,11 +67,11 @@ const processAndIndex = async ({
   transforms: any;
   topicPartition: string;
 }) => {
-  const docs: Array<(typeof transforms)[keyof typeof transforms]["Schema"]> =
-    [];
+  const docs: Array<(typeof transforms)[keyof typeof transforms]["Schema"]> = [];
   for (const kafkaRecord of kafkaRecords) {
     console.log(JSON.stringify(kafkaRecord, null, 2));
     const { value, offset } = kafkaRecord;
+
     try {
       // If a legacy tombstone, continue
       if (!value) {
@@ -75,19 +81,54 @@ const processAndIndex = async ({
       // Parse the kafka record's value
       const record = JSON.parse(decodeBase64WithUtf8(value));
 
+      // query all changelog entries for this ID and create copies of all entries with new ID
+      if (record.isAdminChange) {
+        const schema = transformDeleteSchema(offset).or(
+          transformUpdateValuesSchema(offset).or(transformedUpdateIdSchema),
+        );
+
+        const result = schema.safeParse(record);
+
+        if (result.success) {
+          if (result.data.adminChangeType === "update-id") {
+            docs.forEach((log) => {
+              const recordOffset = log.id.split("-").at(-1);
+
+              docs.push({
+                ...log,
+                id: `${result.data.id}-${recordOffset}`,
+                packageId: result.data.id,
+              });
+            });
+            const packageChangelogs = await getPackageChangelog(result.data.idToBeUpdated);
+
+            packageChangelogs.hits.hits.forEach((log) => {
+              const recordOffset = log._id.split("-").at(-1);
+              docs.push({
+                ...log._source,
+                id: `${result.data.id}-${recordOffset}`,
+                packageId: result.data.id,
+              });
+            });
+          } else {
+            docs.push(result.data);
+          }
+        } else {
+          console.log(
+            `Skipping package with invalid format for type "${record.adminChangeType}"`,
+            result.error.message,
+          );
+        }
+      }
       // If we're not a mako event, continue
       // TODO:  handle legacy.  for now, just continue
       if (!record.event || record?.origin !== "mako") {
         continue;
       }
 
       // If the event is a supported event, transform and push to docs array for indexing
-      console.log("event below");
-      console.log(record.event);
-
       if (record.event in transforms) {
-        const transformForEvent =
-          transforms[record.event as keyof typeof transforms];
+        const transformForEvent = transforms[record.event as keyof typeof transforms];
 
         const result = transformForEvent.transform(offset).safeParse(record);
 

lib/lambda/sinkMain.ts

@@ -5,6 +5,11 @@ import { ErrorType, bulkUpdateDataWrapper, getTopic, logError } from "../libs/si
 import { Index } from "shared-types/opensearch";
 import { decodeBase64WithUtf8 } from "shared-utils";
 import * as os from "./../libs/opensearch-lib";
+import {
+  deleteAdminChangeSchema,
+  updateValuesAdminChangeSchema,
+  updateIdAdminChangeSchema,
+} from "./update/adminChangeSchemas";
 
 const osDomain = process.env.osDomain;
 const indexNamespace = process.env.indexNamespace;
@@ -81,6 +86,22 @@ const processAndIndex = async ({
 
       // If we're not a mako event, continue
       // TODO:  handle legacy.  for now, just continue
+
+      const schema = deleteAdminChangeSchema
+        .or(updateValuesAdminChangeSchema)
+        .or(updateIdAdminChangeSchema);
+
+      if (record.isAdminChange) {
+        const result = schema.safeParse(record);
+        if (result.success) {
+          docs.push(record);
+        } else {
+          console.log(
+            `Skipping package with invalid format for type "${record.adminChangeType}"`,
+            result.error.message,
+          );
+        }
+      }
       if (!record.event || record?.origin !== "mako") {
         continue;
       }

lib/lambda/update/adminChangeSchemas.ts

@@ -0,0 +1,50 @@
+import { z } from "zod";
+
+export const deleteAdminChangeSchema = z
+  .object({
+    id: z.string(),
+    deleted: z.boolean(),
+    adminChangeType: z.literal("delete"),
+  })
+  .and(z.record(z.string(), z.any()));
+
+export const updateValuesAdminChangeSchema = z
+  .object({
+    id: z.string(),
+    adminChangeType: z.literal("update-values"),
+  })
+  .and(z.record(z.string(), z.any()));
+
+export const updateIdAdminChangeSchema = z
+  .object({
+    id: z.string(),
+    adminChangeType: z.literal("update-id"),
+    idToBeUpdated: z.string(),
+  })
+  .and(z.record(z.string(), z.any()));
+
+export const transformDeleteSchema = (offset: number) =>
+  deleteAdminChangeSchema.transform((data) => ({
+    ...data,
+    event: "delete",
+    packageId: data.id,
+    id: `${data.id}-${offset}`,
+    timestamp: Date.now(),
+  }));
+
+export const transformUpdateValuesSchema = (offset: number) =>
+  updateValuesAdminChangeSchema.transform((data) => ({
+    ...data,
+    event: "update-values",
+    packageId: data.id,
+    id: `${data.id}-${offset}`,
+    timestamp: Date.now(),
+  }));
+
+export const transformedUpdateIdSchema = updateIdAdminChangeSchema.transform((data) => ({
+  ...data,
+  event: "update-id",
+  packageId: data.id,
+  id: `${data.id}`,
+  timestamp: Date.now(),
+}));

lib/lambda/update/getPackageType.ts

@@ -0,0 +1,25 @@
+import { response } from "lib/libs/handler-lib";
+import { events } from "lib/packages/shared-types";
+import { getPackageChangelog } from "lib/libs/api/package";
+
+export const getPackageType = async (packageId: string) => {
+  // use event of current package to determine how ID should be formatted
+  try {
+    const packageChangelog = await getPackageChangelog(packageId);
+    if (!packageChangelog.hits.hits.length) {
+      throw new Error("The type of package could not be determined.");
+    }
+
+    const packageWithSubmissionType = packageChangelog.hits.hits.find((pkg) => {
+      return pkg._source.event in events;
+    });
+    const packageEvent = packageWithSubmissionType?._source.event;
+
+    return packageEvent;
+  } catch (error) {
+    return response({
+      statusCode: 500,
+      body: { message: error },
+    });
+  }
+};