feat: prepare organization enrichment and cache tables (CM-685) (#3422)

Author: skwowet

backend/src/database/migrations/U1758292471__organizationEnrichment.sql

@@ -0,0 +1,2 @@
+drop table "organizationEnrichments";
+drop table "organizationEnrichmentCache";

backend/src/database/migrations/V1758292471__organizationEnrichment.sql

@@ -0,0 +1,14 @@
+create table "organizationEnrichments"(
+    "organizationId" uuid primary key,
+    "lastTriedAt"   timestamp with time zone,
+    "lastUpdatedAt" timestamp with time zone
+);
+
+create table "organizationEnrichmentCache" (
+    "organizationId" uuid not null references organizations(id) on delete no action on update no action,
+    "data" jsonb not null,
+    "source" text not null,
+    "createdAt" timestamp with time zone not null,
+    "updatedAt" timestamp with time zone not null,
+    primary key ("organizationId", source)
+);

services/libs/data-access-layer/src/enrichment/organizationEnrichmentCache/index.ts

@@ -0,0 +1,40 @@
+import { IOrganizationEnrichmentCache, OrganizationEnrichmentSource } from '@crowd/types'
+
+import { QueryExecutor } from '../../queryExecutor'
+
+export async function insertOrganizationEnrichmentCache<T>(
+  qx: QueryExecutor,
+  organizationId: string,
+  data: T,
+  source: OrganizationEnrichmentSource,
+): Promise<void> {
+  await qx.result(
+    `insert into "organizationEnrichmentCache" ("organizationId", "data", "source", "createdAt", "updatedAt")
+        values ($(organizationId), $(data), $(source), now(), now())
+        on conflict ("organizationId", "source") do update set "data" = $(data), "updatedAt" = now()`,
+    { organizationId, data, source },
+  )
+}
+
+export async function findOrganizationEnrichmentCache<T>(
+  qx: QueryExecutor,
+  organizationId: string,
+  source: OrganizationEnrichmentSource,
+): Promise<IOrganizationEnrichmentCache<T> | null> {
+  return qx.selectOneOrNone(
+    `select * from "organizationEnrichmentCache" where "organizationId" = $(organizationId) and "source" = $(source)`,
+    { organizationId, source },
+  )
+}
+
+export async function updateOrganizationEnrichmentCache<T>(
+  qx: QueryExecutor,
+  organizationId: string,
+  data: T,
+  source: OrganizationEnrichmentSource,
+): Promise<void> {
+  await qx.result(
+    `update "organizationEnrichmentCache" set "data" = $(data), "updatedAt" = now() where "organizationId" = $(organizationId) and "source" = $(source)`,
+    { organizationId, data, source },
+  )
+}

services/libs/types/src/enrichment.ts

@@ -1,4 +1,4 @@
-import { MemberEnrichmentSource } from './enums'
+import { MemberEnrichmentSource, OrganizationEnrichmentSource } from './enums'
 import { IMemberIdentity, IMemberReach } from './members'
 import { IOrganizationIdentity } from './organizations'
 
@@ -55,3 +55,11 @@ export interface IMemberOriginalData {
   // memberOrganizations table data
   organizations: IMemberOrganizationData[]
 }
+
+export interface IOrganizationEnrichmentCache<T> {
+  createdAt: string
+  updatedAt: string
+  organizationId: string
+  data: T
+  source: OrganizationEnrichmentSource
+}

services/libs/types/src/enums/enrichment.ts

@@ -17,3 +17,7 @@ export enum MemberEnrichmentMaterializedView {
   ENTITY_UPDATES_ANALYSIS = 'memberEnrichmentMonitoringEntityUpdates',
   LLM_USAGE_ANALYSIS = 'memberEnrichmentMonitoringLLMQueries',
 }
+
+export enum OrganizationEnrichmentSource {
+  INTERNAL_API = 'internal-api',
+}