diff --git a/analytics/src/features.test.ts b/analytics/src/features.test.ts index f7813d9..d15d763 100644 --- a/analytics/src/features.test.ts +++ b/analytics/src/features.test.ts @@ -31,6 +31,12 @@ function makeRedis() { }; }), zcard: vi.fn(async (key: string) => zsets.get(key)?.size ?? 0), + zadd: vi.fn(async (key: string, score: number, member: string) => { + const set = zsets.get(key) ?? new Map(); + set.set(member, score); + zsets.set(key, set); + return 0; + }), zremrangebyrank: vi.fn(async () => 0), zscore: vi.fn(async (key: string, member: string) => { const v = zsets.get(key)?.get(member); @@ -53,6 +59,20 @@ describe("FeatureMaterializer", () => { expect(zsets.get("features:author:pos:viewer-1")?.get("author-1")).toBe(5); }); + it("adjusts positive evidence for unliked/unfollowed and clamps to non-negative", async () => { + const { redis, zsets } = makeRedis(); + const features = new FeatureMaterializer(redis); + await features.recordPositive("viewer-1", "author-1", ["rust"], 2); + expect(zsets.get("features:author:pos:viewer-1")?.get("author-1")).toBe(2); + + await features.recordPositiveAdjustment("viewer-1", "author-1", ["rust"], -2); + expect(zsets.get("features:author:pos:viewer-1")?.get("author-1")).toBe(0); + + // Over-reduction clamps at 0 + await features.recordPositiveAdjustment("viewer-1", "author-1", ["rust"], -2); + expect(zsets.get("features:author:pos:viewer-1")?.get("author-1")).toBe(0); + }); + it("keeps hide evidence separate from positive affinity", async () => { const { redis, zsets } = makeRedis(); const features = new FeatureMaterializer(redis); diff --git a/analytics/src/features.ts b/analytics/src/features.ts index 6631ca0..ce140b6 100644 --- a/analytics/src/features.ts +++ b/analytics/src/features.ts @@ -42,6 +42,25 @@ export class FeatureMaterializer { await this.trim(`features:topic:pos:${viewerId}`); } + /** Reversal adjustments: reduce positive evidence when a like or follow is undone. */ + async recordPositiveAdjustment(viewerId: string, authorId: string | null, topics: string[], delta: number): Promise { + if (delta >= 0) return; + const pipeline = this.redis.pipeline(); + if (authorId) { + pipeline.zincrby(`features:author:pos:${viewerId}`, delta, authorId); + pipeline.expire(`features:author:pos:${viewerId}`, AFFINITY_TTL_SECONDS); + } + for (const topic of topics.slice(0, 10)) { + pipeline.zincrby(`features:topic:pos:${viewerId}`, delta, topic.toLowerCase()); + pipeline.expire(`features:topic:pos:${viewerId}`, AFFINITY_TTL_SECONDS); + } + await pipeline.exec(); + await this.clampNonNegative(`features:author:pos:${viewerId}`, authorId); + for (const topic of topics.slice(0, 10)) { + await this.clampNonNegative(`features:topic:pos:${viewerId}`, topic.toLowerCase()); + } + } + /** Hide evidence: separate negative counters, never subtracted embeddings. */ async recordHide(viewerId: string, authorId: string | null, topics: string[]): Promise { const pipeline = this.redis.pipeline(); @@ -92,6 +111,14 @@ export class FeatureMaterializer { return (await this.redis.sismember(`features:seen:${viewerId}`, postId)) === 1; } + private async clampNonNegative(key: string, member: string | null): Promise { + if (!member) return; + const score = await this.redis.zscore(key, member); + if (score !== null && Number.parseFloat(score) < 0) { + await this.redis.zadd(key, 0, member); + } + } + private async trim(key: string): Promise { const size = await this.redis.zcard(key); if (size > TOP_N) { diff --git a/analytics/src/processor.test.ts b/analytics/src/processor.test.ts index 8950c78..891c893 100644 --- a/analytics/src/processor.test.ts +++ b/analytics/src/processor.test.ts @@ -170,4 +170,38 @@ describe("AnalyticsProcessor", () => { expect(trending.record).not.toHaveBeenCalled(); expect(batch.resolveOffset).toHaveBeenCalledWith("0"); }); + + it("materializes features for post.unliked, user.unfollowed, post.dwell, and media.progress", async () => { + const { redis, trending } = makeDeps(); + const recordPositive = vi.fn().mockResolvedValue(undefined); + const recordPositiveAdjustment = vi.fn().mockResolvedValue(undefined); + const features = { + recordSeen: vi.fn(), + recordPositive, + recordPositiveAdjustment, + recordHide: vi.fn(), + }; + const processor = new AnalyticsProcessor( + {} as never, + redis as never, + { groupId: "test", topics: ["behavior.events.v2"] }, + trending as never, + new (await import("./receipts.js")).ReceiptStore(redis as never), + features as never + ); + + const events = [ + envelope("evt-unliked", { eventType: "post.unliked", payload: { postId: "post-1", authorId: "author-1", tags: ["tech"] } }), + envelope("evt-unfollowed", { eventType: "user.unfollowed", payload: { followeeId: "author-2" } }), + envelope("evt-dwell", { eventType: "post.dwell", payload: { postId: "post-1", activeDwellMs: 12000, authorId: "author-1" } }), + envelope("evt-progress", { eventType: "media.progress", payload: { postId: "post-1", progress: 0.8, authorId: "author-1" } }), + ]; + + await processor.handleBatch(makeBatch(events)); + + expect(recordPositiveAdjustment).toHaveBeenCalledWith("user-1", "author-1", ["tech"], -2); + expect(recordPositiveAdjustment).toHaveBeenCalledWith("user-1", "author-2", [], -4); + expect(recordPositive).toHaveBeenCalledWith("user-1", "author-1", [], 1); + expect(recordPositive).toHaveBeenCalledWith("user-1", "author-1", [], 1); + }); }); diff --git a/analytics/src/processor.ts b/analytics/src/processor.ts index fd0d7a9..8ef6bb2 100644 --- a/analytics/src/processor.ts +++ b/analytics/src/processor.ts @@ -85,6 +85,8 @@ export class AnalyticsProcessor { } const contribution = trendingContribution(parsed.eventType); if (contribution.kind === null) { + // Event type doesn't contribute to trending, but might still materialize features (e.g. dwell, progress, unliked) + await this.materializeFeatures(parsed); payload.resolveOffset(message.offset); continue; } @@ -146,12 +148,32 @@ export class AnalyticsProcessor { case "post.liked": await this.features.recordPositive(viewerId, authorId, tags, 2); break; + case "post.unliked": + await this.features.recordPositiveAdjustment(viewerId, authorId, tags, -2); + break; case "post.commented": await this.features.recordPositive(viewerId, authorId, tags, 3); break; case "user.followed": await this.features.recordPositive(viewerId, authorId, [], 4); break; + case "user.unfollowed": + await this.features.recordPositiveAdjustment(viewerId, authorId, [], -4); + break; + case "post.dwell": { + const activeDwellMs = typeof parsed.payload.activeDwellMs === "number" ? parsed.payload.activeDwellMs : 0; + if (activeDwellMs >= 10000) { + await this.features.recordPositive(viewerId, authorId, tags, 1); + } + break; + } + case "media.progress": { + const progress = typeof parsed.payload.progress === "number" ? parsed.payload.progress : 0; + if (progress >= 0.5) { + await this.features.recordPositive(viewerId, authorId, tags, 1); + } + break; + } case "post.hidden": await this.features.recordHide(viewerId, authorId, tags); break; diff --git a/analytics/src/suppression.test.ts b/analytics/src/suppression.test.ts new file mode 100644 index 0000000..c73bdec --- /dev/null +++ b/analytics/src/suppression.test.ts @@ -0,0 +1,45 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +import { SuppressionStore } from "./suppression.js"; + +function makeRedis() { + const sets = new Map>(); + const redis = { + sadd: vi.fn(async (key: string, member: string) => { + const set = sets.get(key) ?? new Set(); + set.add(member); + sets.set(key, set); + return 1; + }), + sismember: vi.fn(async (key: string, member: string) => (sets.get(key)?.has(member) ? 1 : 0)), + }; + return { redis: redis as unknown as import("ioredis").Redis, sets }; +} + +describe("SuppressionStore", () => { + beforeEach(() => vi.clearAllMocks()); + + it("suppresses users and checks suppression status correctly", async () => { + const { redis } = makeRedis(); + const suppression = new SuppressionStore(redis); + + expect(await suppression.isUserSuppressed("user-100")).toBe(false); + + await suppression.suppressUser("user-100"); + + expect(await suppression.isUserSuppressed("user-100")).toBe(true); + expect(await suppression.isUserSuppressed("user-200")).toBe(false); + }); + + it("suppresses posts and checks suppression status correctly", async () => { + const { redis } = makeRedis(); + const suppression = new SuppressionStore(redis); + + expect(await suppression.isPostSuppressed("post-500")).toBe(false); + + await suppression.suppressPost("post-500"); + + expect(await suppression.isPostSuppressed("post-500")).toBe(true); + expect(await suppression.isPostSuppressed("post-600")).toBe(false); + }); +});