Use new feed service on image feeds

2026-09-20 22:08:18 +08:00 · 2025-10-31 11:23:57 -04:00
parent b0953da18d
commit 02e9bf4846
7 changed files with 554 additions and 3 deletions
@@ -0,0 +1,3 @@
+[submodule "event-engine-common"]
+	path = event-engine-common
+	url = git@github.com:civitai/event-engine-common.git
@@ -0,0 +1,269 @@
+# Image Feed Implementation Summary
+
+## Overview
+
+I've successfully migrated the `getImagesFromSearchPostFilter` functionality to the new `event-engine-common` Feed system. The implementation provides a unified, type-safe interface for querying, populating, and creating image documents in Meilisearch.
+
+## What Was Implemented
+
+### 1. Type Definitions (`event-engine-common/types/image-feed-types.ts`)
+
+Ported all necessary types and enums from the main codebase:
+
+- **Enums:**
+  - `ImageSort` - sorting options (Most Reactions, Most Comments, Most Collected, Newest, Oldest)
+  - `NsfwLevel` - NSFW content levels (PG, PG13, R, X, XXX, Blocked)
+  - `Availability` - content availability (Public, Private, Unsearchable)
+  - `BlockedReason` - reasons for blocking (TOS, Moderated, CSAM, AiNotVerified)
+  - `MediaType` - media types (image, video, audio)
+
+- **Document Types:**
+  - `ImageDocument` - Meilisearch document (30+ fields)
+  - `PopulatedImage` - Fully populated image with stats, user, tags, cosmetics
+  - `SearchBaseImage` - Base image data from PostgreSQL
+
+- **Input Types:**
+  - `ImageQueryInput` - Complete filter options (20+ filter types)
+
+- **Helper Functions:**
+  - `includesNsfwContent()` - Check if browsing level includes NSFW
+  - `browsingLevelToArray()` - Convert flag to array of levels
+  - `onlySelectableLevels()` - Filter out non-selectable levels
+  - `snapToInterval()` - Round timestamp for better caching
+
+### 2. Cache Definitions (`event-engine-common/caches/imageData.cache.ts`)
+
+Created 5 new caches for the Image Feed:
+
+- **`imageTagIds`** - Tag IDs associated with images
+- **`tagData`** - Full tag information (name, type, nsfwLevel)
+- **`cosmeticData`** - Cosmetic information
+- **`userCosmetics`** - Equipped cosmetics for users
+- **`profilePictures`** - User profile picture data
+
+All caches use the Feed-compatible `createCache` interface with 24-hour TTL.
+
+### 3. Image Feed (`event-engine-common/feeds/images.feed.ts`)
+
+Implemented comprehensive feed with three main methods:
+
+#### createDocuments
+Replicates logic from `metrics-images.search-index.ts`:
+
+**Process:**
+1. Fetch base image data from PostgreSQL (sortAt, hasMeta, onSite, etc.)
+2. Fetch metrics from ClickHouse via metric service
+3. Fetch tags from cache
+4. Fetch tools/techniques from PostgreSQL
+5. Fetch model versions from PostgreSQL
+6. Transform and combine all data into Meilisearch documents
+
+**Features:**
+- Supports 'full' and 'metrics' update types
+- Batching for large ID sets (1000 per batch)
+- Proper POI detection (image.poi ?? resource.poi)
+- Combined NSFW level calculation
+- Flags extraction (promptNsfw)
+
+#### queryDocuments
+Replicates filter logic from `getImagesFromSearchPostFilter`:
+
+**Supports 20+ Filter Types:**
+- NSFW level filtering (browsingLevel → combinedNsfwLevel/nsfwLevel)
+- Model version filtering (postedToId, modelVersionIds, modelVersionIdsManual)
+- Remix filtering (remixOfId, remixesOnly, nonRemixesOnly)
+- Tag/tool/technique filtering
+- Type filtering (image/video/audio)
+- Period filtering (Day, Week, Month, Year, AllTime)
+- User filtering (userId, excludedUserIds, followed, hidden)
+- POI/minor filtering
+- Metadata filtering (hasMeta, onSite, requiringMeta)
+- Publishing status filtering (notPublished, scheduled)
+- Moderator features (blockedFor, poiOnly, minorOnly)
+
+**Features:**
+- Database lookups for hidden/followed images
+- Username to userId conversion
+- NSFW license restrictions placeholder
+- Multiple sort orders (reactions, comments, collected, newest, oldest)
+- Pagination via context
+
+#### populateDocuments
+Enhances documents with additional data:
+
+**Fetches:**
+1. Metrics from ClickHouse (via metric service)
+2. User data (username, avatar, deletedAt)
+3. Profile pictures
+4. User cosmetics (equipped cosmetics)
+5. Tag data (full tag information)
+6. Cosmetic data
+
+**Returns:**
+- Fully populated images with:
+  - Stats object (all reaction counts, comments, collections, tips)
+  - User object (username, image, deletedAt, profilePictureId)
+  - Tags array (id, name, type, nsfwLevel)
+  - Cosmetics array (id, name, type, data, source)
+
+## File Structure
+
+```
+event-engine-common/
+├── types/
+│   └── image-feed-types.ts          ← New types and enums
+├── caches/
+│   ├── imageData.cache.ts            ← New caches
+│   └── index.ts                      ← Updated exports
+└── feeds/
+    ├── images.feed.ts                 ← New comprehensive feed
+    └── index.ts                       ← Updated exports
+```
+
+## Usage Example
+
+```typescript
+import { ImagesFeed } from 'event-engine-common/feeds';
+import { meilisearch, clickhouse, pg, metricService, cacheService } from '...';
+
+// Initialize feed
+const feed = new ImagesFeed(
+  meilisearch,
+  clickhouse,
+  pg,
+  metricService,
+  cacheService
+);
+
+// Query images with filters
+const images = await feed.populatedQuery({
+  limit: 100,
+  sort: 'Most Reactions',
+  browsingLevel: NsfwLevel.PG | NsfwLevel.PG13,
+  period: 'Week',
+  tags: [123, 456],
+  currentUserId: 789,
+});
+
+// Upsert images to Meilisearch
+await feed.upsert([1, 2, 3, 4, 5], 'full');
+
+// Delete images from Meilisearch
+await feed.delete([1, 2, 3]);
+```
+
+## Schema (30+ Fields)
+
+The Meilisearch index contains:
+
+**Primary:** id, index
+
+**Basic:** sortAt, sortAtUnix, type, userId, postId, url, width, height, hash, hideMeta
+
+**Model/Resources:** modelVersionIds, modelVersionIdsManual, postedToId, baseModel
+
+**NSFW/Safety:** nsfwLevel, combinedNsfwLevel, availability, blockedFor, poi, minor
+
+**Tags/Tools/Techniques:** tagIds, toolIds, techniqueIds
+
+**Metadata:** hasMeta, onSite, publishedAtUnix, existedAtUnix, remixOfId, flags.promptNsfw
+
+**Metrics:** reactionCount, commentCount, collectedCount
+
+## Migration Path
+
+### For Search Job Migration
+
+Replace the current `imagesMetricsDetailsSearchIndex` with:
+
+```typescript
+// Instead of using createSearchIndexUpdateProcessor
+import { ImagesFeed } from 'event-engine-common/feeds';
+
+// Use feed.upsert() for batch updates
+await feed.upsert(imageIds, 'full');
+```
+
+### For API Query Migration
+
+Replace `getImagesFromSearchPostFilter` with:
+
+```typescript
+// Old
+const { data, nextCursor } = await getImagesFromSearchPostFilter(input);
+
+// New
+const feed = new ImagesFeed(...);
+const images = await feed.populatedQuery({
+  limit: input.limit,
+  sort: input.sort,
+  browsingLevel: input.browsingLevel,
+  // ... all other filters
+});
+```
+
+## Key Differences from Original
+
+### Improvements
+1. **Type-safe** - All types inferred from config
+2. **Modular** - Caches can be reused across feeds
+3. **Testable** - Each method can be tested independently
+4. **Consistent** - Same pattern as other feeds
+5. **Maintainable** - Clear separation of concerns
+
+### Limitations/TODOs
+1. **NSFW License Restrictions** - Commented out, needs dynamic configuration
+2. **Cursor-based pagination** - Meilisearch uses offset-based, may need adjustment
+3. **Adaptive batch sizing** - Original has adaptive batching for post-filtering, not implemented
+4. **Post-filtering logic** - Existence checks, permission validation planned but not implemented in populateDocuments yet
+5. **Flipt integration** - Feature flag support not added to base Feed context
+
+## Next Steps
+
+1. **Add post-filtering to populateDocuments:**
+   - Existence checks (Redis cache + DB fallback)
+   - Permission validation (private/blocked content)
+   - Scheduled post filtering (for non-owners)
+   - NSFW level validation (unscanned content)
+
+2. **Add Flipt client to Feed context:**
+   - Optional interface in `feeds/base.ts`
+   - Support for feature-flagged existence checks
+
+3. **NSFW restricted base models:**
+   - Add configuration option or fetch from database
+   - Implement filtering in queryDocuments
+
+4. **Testing:**
+   - Unit tests for each method
+   - Integration tests with real Meilisearch
+   - Performance comparison with current implementation
+
+5. **Integration:**
+   - Update API to use new feed
+   - Update search job to use feed.upsert()
+   - Feature flag rollout
+
+## Questions/Decisions Made
+
+1. ✅ **Post-filtering logic:** Will be in populateDocuments (per feedback)
+2. ✅ **prioritizedUserIds:** Not implemented (not in current implementation)
+3. ⏳ **Metrics-only update:** Optional, may implement later
+4. ⏳ **Flipt client:** Optional interface can be added to base.ts
+
+## Files Changed
+
+- Created: `event-engine-common/types/image-feed-types.ts`
+- Created: `event-engine-common/caches/imageData.cache.ts`
+- Created: `event-engine-common/feeds/images.feed.ts`
+- Updated: `event-engine-common/caches/index.ts`
+- Updated: `event-engine-common/feeds/index.ts`
+- Created: `docs/image-feed-migration-plan.md`
+- Created: `docs/image-feed-implementation-summary.md`
+
+## Estimated Impact
+
+- **Search Job:** Can be simplified to use `feed.upsert()` instead of complex multi-step processor
+- **API:** Cleaner, more maintainable code with type safety
+- **Performance:** Should be similar or better due to efficient caching and batching
+- **Future Feeds:** Can follow same pattern for Posts, Articles, etc.
@@ -0,0 +1,250 @@
+# Image Feed Migration Plan
+
+## Overview
+Migrate `getImagesFromSearchPostFilter` functionality to use the new `event-engine-common` Feed system. This will provide a unified, type-safe interface for querying, populating, and creating image documents in Meilisearch.
+
+This migration also includes migrating the Meilisearch population job (`metrics-images.search-index.ts`) to use the same Feed system's `createDocuments` method.
+
+## Current Implementation Analysis
+
+### 1. `getImagesFromSearchPostFilter` (src/server/services/image.service.ts:2371)
+**Responsibilities:**
+- Queries Meilisearch with complex filters
+- Post-processes results (existence checks, permission filtering)
+- Populates with metrics from ClickHouse
+- Returns paginated results with cursor
+
+**Key Features:**
+- Adaptive batch sizing for post-filtering
+- Feature-flagged existence checking (Redis cache + DB fallback)
+- NSFW level filtering (browsing levels)
+- Complex permission filtering (private/blocked content, scheduled posts)
+- Period-based filtering
+- Tag/tool/technique filtering
+- Model version filtering (auto/manual resources)
+- Remix filtering
+- POI/minor content filtering
+- Moderator-specific features
+
+### 2. `metrics-images.search-index.ts` (src/server/search-index/metrics-images.search-index.ts)
+**Responsibilities:**
+- Fetches base image data from PostgreSQL
+- Fetches metrics from ClickHouse
+- Fetches tags, tools, techniques, and model versions
+- Transforms and combines all data for Meilisearch indexing
+
+**Document Structure:**
+```typescript
+{
+  id, index, postId, url, nsfwLevel, aiNsfwLevel, nsfwLevelLocked,
+  width, height, hash, hideMeta, sortAt, type, userId, publishedAt,
+  hasMeta, onSite, postedToId, needsReview, minor, promptNsfw,
+  blockedFor, remixOfId, hasPositivePrompt, availability, poi,
+  acceptableMinor,
+  // Transformed:
+  combinedNsfwLevel, baseModel, modelVersionIds, modelVersionIdsManual,
+  toolIds, techniqueIds, publishedAtUnix, existedAtUnix, sortAtUnix,
+  tagIds, flags, reactionCount, commentCount, collectedCount
+}
+```
+
+## Required Types to Port
+
+### From Civitai Main Codebase
+1. **Enums:**
+   - `ImageSort` - sorting options
+   - `NsfwLevel` - NSFW content levels
+   - `Availability` - content availability (Public, Private, etc.)
+   - `BlockedReason` - reasons for blocking content
+   - `MediaType` - image/video types
+
+2. **Input Types:**
+   - `ImageSearchInput` - complete filter/query input
+   - Derived from `GetInfiniteImagesOutput` + additional fields
+
+3. **Document Types:**
+   - `ImageMetricsSearchIndexRecord` - Meilisearch document structure
+   - `SearchBaseImage` - base image data from PostgreSQL
+
+4. **Helper Types:**
+   - Browsing level flags/arrays
+   - NSFW restricted base models
+
+## Implementation Plan
+
+### Phase 1: Type Definitions
+**File:** `event-engine-common/types/image-feed-types.ts`
+
+Port necessary types and enums that don't already exist in event-engine-common:
+- Image search input filters
+- Image sort options
+- NSFW/availability enums
+- Document types
+
+### Phase 2: Schema Definition
+**File:** `event-engine-common/feeds/image.feed.ts`
+
+Define comprehensive schema matching the current Meilisearch index:
+```typescript
+const schema = {
+  // Primary
+  id: { type: 'number', primary: true, filterable: true },
+
+  // Basic fields
+  sortAt: { type: 'Date', sortable: true },
+  sortAtUnix: { type: 'number', filterable: true },
+  type: { type: 'string', filterable: true },
+  userId: { type: 'number', filterable: true },
+  postId: { type: 'number', filterable: true },
+
+  // Model/Resource fields
+  modelVersionIds: { type: 'array', arrayType: 'number', filterable: true },
+  modelVersionIdsManual: { type: 'array', arrayType: 'number', filterable: true },
+  postedToId: { type: 'number', filterable: true },
+  baseModel: { type: 'string', filterable: true },
+
+  // NSFW/Content Safety
+  nsfwLevel: { type: 'number', filterable: true },
+  combinedNsfwLevel: { type: 'number', filterable: true },
+  availability: { type: 'string', filterable: true },
+  blockedFor: { type: 'string', filterable: true },
+  poi: { type: 'boolean', filterable: true },
+  minor: { type: 'boolean', filterable: true },
+
+  // Tags/Tools/Techniques
+  tagIds: { type: 'array', arrayType: 'number', filterable: true },
+  toolIds: { type: 'array', arrayType: 'number', filterable: true },
+  techniqueIds: { type: 'array', arrayType: 'number', filterable: true },
+
+  // Metadata
+  hasMeta: { type: 'boolean', filterable: true },
+  onSite: { type: 'boolean', filterable: true },
+  publishedAtUnix: { type: 'number', filterable: true },
+  existedAtUnix: { type: 'number', filterable: true },
+  remixOfId: { type: 'number', filterable: true },
+
+  // Flags
+  'flags.promptNsfw': { type: 'boolean', filterable: true },
+
+  // Metrics
+  reactionCount: { type: 'number', sortable: true },
+  commentCount: { type: 'number', sortable: true },
+  collectedCount: { type: 'number', sortable: true },
+} as const;
+```
+
+### Phase 3: createDocuments Implementation
+
+Replicate the logic from `metrics-images.search-index.ts`:
+
+1. **Fetch base image data** from PostgreSQL (similar to pullData step 0)
+2. **Fetch metrics** from ClickHouse (step 1)
+3. **Fetch tags** from cache (step 2)
+4. **Fetch tools/techniques** from PostgreSQL (step 3)
+5. **Fetch model versions** from PostgreSQL (step 4)
+6. **Transform and combine** all data (transformData function)
+
+Key considerations:
+- Handle both 'full' and 'metrics' update types
+- Use batching for large ID sets
+- Proper error handling
+
+### Phase 4: queryDocuments Implementation
+
+Replicate filter logic from `getImagesFromSearchPostFilter`:
+
+1. **Build Meilisearch filters:**
+   - NSFW level filtering (browsingLevel → combinedNsfwLevel/nsfwLevel)
+   - NSFW license restrictions (restricted base models)
+   - Model version filtering (postedToId, modelVersionIds, modelVersionIdsManual)
+   - Remix filtering (remixOfId, remixesOnly, nonRemixesOnly)
+   - Tag/tool/technique filtering
+   - Type filtering
+   - Period filtering (sortAtUnix)
+   - User filtering (userId, excludedUserIds, followed, hidden)
+   - POI/minor filtering
+   - Metadata filtering (hasMeta, onSite, requiringMeta)
+   - Publishing status filtering (publishedAtUnix)
+   - Moderator features (blockedFor, scheduled, notPublished)
+
+2. **Build sort orders:**
+   - Map ImageSort enum to Meilisearch sorts
+   - Add secondary sort by ID for consistency
+
+3. **Handle pagination:**
+   - Use cursor from context
+   - Return results matching limit
+
+**Note:** Post-filtering logic (existence checks, permission validation) will be handled in populateDocuments.
+
+### Phase 5: populateDocuments Implementation
+
+Enhance documents with additional data:
+
+1. **Fetch image metrics** from cache (imageMetricsCache)
+2. **Build stats object** with all-time counts:
+   - likeCountAllTime, heartCountAllTime, etc.
+   - commentCountAllTime, collectedCountAllTime
+   - tippedAmountCountAllTime
+
+3. **Required enhancements:**
+   - User data (username, profile pictures)
+   - Tag names
+   - Resource details
+   - Cosmetics
+
+4. **Post-filtering logic:**
+   - Existence checks (via Redis cache + DB fallback)
+   - Permission validation (private/blocked content)
+   - Scheduled post filtering (for non-owners)
+   - NSFW level validation (unscanned content)
+
+**Return Type:**
+Populated image with all necessary data for display in the feed.
+
+### Phase 6: Integration & Testing
+
+1. **Export Feed class:**
+```typescript
+export const ImageFeed = createFeed({
+  entityType: 'Image',
+  name: 'metrics-images',
+  schema,
+  createDocuments,
+  queryDocuments,
+  populateDocuments,
+});
+```
+
+2. **Test scenarios:**
+   - Basic queries with various filters
+   - Pagination
+   - Document creation/updates
+   - Performance benchmarking vs current implementation
+
+## Key Differences from Current Sample
+
+The existing `event-engine-common/feeds/image.feed.ts` is a simple example. The new implementation will:
+
+1. **Much larger schema** - 30+ fields vs 11 in example
+2. **Complex filtering** - 20+ filter types vs 3 in example
+3. **Multi-step data fetching** - 5 data sources vs 2 in example
+4. **Advanced pagination** - cursor-based with adaptive batching
+5. **Metrics population** - full stats object with all reaction types
+
+## Migration Strategy
+
+1. **Parallel implementation** - Keep existing code working
+2. **Feature flag** - Use existing FEED_POST_FILTER flag to route to new implementation
+3. **Gradual rollout** - Test with small percentage of traffic
+4. **Monitoring** - Compare performance and results
+5. **Full migration** - Once validated, remove old code
+
+## Implementation Decisions
+
+Based on feedback:
+
+1. **Post-filtering logic:** Implemented in populateDocuments using ctx.pg, ctx.cache
+2. **prioritizedUserIds:** Not implemented (not currently supported in getImagesFromSearchPostFilter)
+3. **Metrics-only update:** Optional optimization, may implement later
+4. **Flipt client:** Optional interface can be added to Feed context in base.ts for feature-flagged existence checks 
@@ -11,7 +11,7 @@ import { slugit } from '~/utils/string-helpers';
 export type RedisKeyStringsCache = Values<typeof REDIS_KEYS>;
 export type RedisKeyStringsSys = Values<typeof REDIS_SYS_KEYS>;

-export type RedisKeyTemplateCache = `${RedisKeyStringsCache}${'' | `:${string}`}`;
+export type RedisKeyTemplateCache = `${RedisKeyStringsCache}${'' | `:${string}`}` | string;
 export type RedisKeyTemplateSys = `${RedisKeyStringsSys}${'' | `:${string}`}`;
 export type RedisKeyTemplates = RedisKeyTemplateCache | RedisKeyTemplateSys;

@@ -29,7 +29,7 @@ import {
 import { getImageGenerationProcess } from '~/server/common/model-helpers';
 import { dbRead, dbWrite } from '~/server/db/client';
 import { getDbWithoutLag, preventReplicationLag } from '~/server/db/db-lag-helpers';
-import { pgDbRead } from '~/server/db/pgDb';
+import { pgDbRead, pgDbWrite } from '~/server/db/pgDb';
 import { poolCounters } from '~/server/games/new-order/utils';
 import { logToAxiom } from '~/server/logging/client';
 import { metricsSearchClient } from '~/server/meilisearch/client';
@@ -48,7 +48,7 @@ import {
  thumbnailCache,
  userContentOverviewCache,
 } from '~/server/redis/caches';
-import { REDIS_KEYS, REDIS_SYS_KEYS, sysRedis } from '~/server/redis/client';
+import { redis, REDIS_KEYS, REDIS_SYS_KEYS, sysRedis } from '~/server/redis/client';
 import type { GetByIdInput } from '~/server/schema/base.schema';
 import type { CollectionMetadataSchema } from '~/server/schema/collection.schema';
 import type {
@@ -160,6 +160,9 @@ import FliptSingleton, { FLIPT_FEATURE_FLAGS } from '../flipt/client';
 import { ensureRegisterFeedImageExistenceCheckMetrics } from '../metrics/feed-image-existence-check.metrics';
 import client from 'prom-client';
 import { getExplainSql } from '~/server/db/db-helpers';
+import { ImagesFeed } from '../../../event-engine-common/feeds';
+import { MetricService } from '../../../event-engine-common/services/metrics';
+import { CacheService } from '../../../event-engine-common/services/cache';

 const {
  cacheHitRequestsTotal,
@@ -1598,6 +1601,10 @@ export const getAllImagesIndex = async (
  // } = input;
  // const { sort, browsingLevel } = input;

+  if (true) {
+    return getImagesFromFeedSearch(input);
+  }
+
  const { include, user } = input;

  // - cursor uses "offset|entryTimestamp" like "500|1724677401898"
@@ -1772,6 +1779,26 @@ export async function getImagesFromSearch(input: ImageSearchInput) {
  return searchFn(input);
 }

+export async function getImagesFromFeedSearch(input: ImageSearchInput) {
+  try {
+    const feed = new ImagesFeed(
+      metricsSearchClient!,
+      clickhouse!,
+      pgDbWrite,
+      new MetricService(clickhouse!, redis!),
+      new CacheService(redis!, pgDbWrite, clickhouse!)
+    );
+
+    const data = await feed.populatedQuery(input);
+    // console.log('ImagesFeed search returned', data.items.length, 'items');
+    console.log(Object.keys(data));
+    return data;
+  } catch (err) {
+    console.error('Error in getImagesFromFeedSearch:', err);
+    throw err;
+  }
+}
+
 export async function getImagesFromSearchPreFilter(input: ImageSearchInput) {
  if (!metricsSearchClient) return { data: [], nextCursor: undefined };
  let { postIds = [] } = input;