memind/plaza-events.mjs

import crypto from 'node:crypto';

const PROFILE_WINDOW_MS = 30 * 24 * 60 * 60 * 1000;
const PROFILE_CACHE_TTL_MS = 45_000;

export const PLAZA_EVENT_TYPES = new Set([
  'impression',
  'click',
  'view',
  'dwell',
  'like',
  'collect',
  'comment',
  'share',
  'dislike',
  'hide',
]);

export const PLAZA_EVENT_WEIGHTS = {
  impression: 0.06,
  click: 0.85,
  view: 1.25,
  dwell: 1.0,
  like: 5.5,
  collect: 6.5,
  comment: 4.5,
  share: 3.5,
  dislike: -9.0,
  hide: -6.0,
};

const DWELL_THRESHOLDS = [
  { ms: 30_000, boost: 4.0 },
  { ms: 10_000, boost: 2.5 },
  { ms: 3_000, boost: 1.5 },
];

function plazaError(message, code) {
  return Object.assign(new Error(message), { code });
}

const DEEP_SEEN_EVENTS = new Set([
  'click',
  'view',
  'dwell',
  'like',
  'collect',
  'comment',
  'share',
]);

function emptyProfile() {
  return {
    profileVersion: 1,
    categoryWeights: {},
    tagWeights: {},
    authorWeights: {},
    feedCategoryWeights: {},
    seenPostIds: new Set(),
    deepSeenPostIds: new Set(),
    dislikedPostIds: new Set(),
    dislikedCategorySlugs: new Set(),
    likedPostIds: new Set(),
    followedAuthorIds: new Set(),
    eventCount: 0,
  };
}

function decayFactor(ageMs, halfLifeMs) {
  if (ageMs <= 0) return 1;
  return Math.pow(0.5, ageMs / halfLifeMs);
}

function bumpWeight(map, key, delta) {
  if (!key) return;
  map[key] = (map[key] ?? 0) + delta;
}

function normalizeWeights(map) {
  const entries = Object.entries(map);
  if (entries.length === 0) return map;
  const max = Math.max(...entries.map(([, value]) => value), 1e-6);
  return Object.fromEntries(entries.map(([key, value]) => [key, value / max]));
}

function dwellBoost(dwellMs) {
  const ms = Number(dwellMs ?? 0);
  if (!Number.isFinite(ms) || ms <= 0) return 0;
  for (const tier of DWELL_THRESHOLDS) {
    if (ms >= tier.ms) return tier.boost;
  }
  return 0;
}

function eventSignal(event) {
  let weight = PLAZA_EVENT_WEIGHTS[event.event_type] ?? 0;
  if (event.event_type === 'dwell') {
    weight *= 1 + dwellBoost(event.dwell_ms);
  }
  return weight;
}

export function createPlazaEventService(pool, { idFactory = () => crypto.randomUUID() } = {}) {
  const profileCache = new Map();

  const cacheKey = ({ userId, sessionId }) =>
    userId ? `u:${userId}` : sessionId ? `s:${sessionId}` : 'anon';

  const invalidateProfileCache = ({ userId, sessionId }) => {
    profileCache.delete(cacheKey({ userId, sessionId }));
    if (userId) profileCache.delete(`u:${userId}`);
    if (sessionId) profileCache.delete(`s:${sessionId}`);
  };

  const recordEvents = async ({
    userId = null,
    sessionId,
    events = [],
  }) => {
    if (!sessionId) throw plazaError('session_id 不能为空', 'invalid_input');
    const now = Date.now();
    const normalized = [];
    for (const raw of events) {
      const eventType = String(raw?.event_type ?? raw?.type ?? '').trim();
      if (!PLAZA_EVENT_TYPES.has(eventType)) continue;
      const postId = String(raw?.post_id ?? '').trim();
      if (!postId) continue;
      normalized.push({
        id: idFactory(),
        userId,
        sessionId,
        postId,
        eventType,
        dwellMs:
          raw?.dwell_ms == null || raw?.dwell_ms === ''
            ? null
            : Math.max(0, Math.min(3_600_000, Number(raw.dwell_ms))),
        feedSort: String(raw?.feed_sort ?? '').slice(0, 32),
        feedCategory: String(raw?.feed_category ?? '').slice(0, 64),
        position:
          raw?.position == null || raw?.position === ''
            ? null
            : Math.max(0, Math.floor(Number(raw.position))),
        createdAt: Number(raw?.created_at) > 0 ? Number(raw.created_at) : now,
      });
    }
    if (normalized.length === 0) {
      return { accepted: 0 };
    }

    const values = normalized.map((event) => [
      event.id,
      event.userId,
      event.sessionId,
      event.postId,
      event.eventType,
      event.dwellMs,
      event.feedSort,
      event.feedCategory,
      event.position,
      event.createdAt,
    ]);
    await pool.query(
      `INSERT INTO plaza_user_events
       (id, user_id, session_id, post_id, event_type, dwell_ms, feed_sort, feed_category, position, created_at)
       VALUES ?`,
      [values],
    );
    invalidateProfileCache({ userId, sessionId });
    return { accepted: normalized.length };
  };

  const buildProfileFromSignals = async ({ userId, sessionId, halfLifeMs }) => {
    const profile = emptyProfile();
    const since = Date.now() - PROFILE_WINDOW_MS;
    const params = [since];
    let identityClause = '';
    if (userId) {
      identityClause = 'AND (e.user_id = ? OR e.session_id = ?)';
      params.push(userId, sessionId);
    } else {
      identityClause = 'AND e.session_id = ?';
      params.push(sessionId);
    }

    const [eventRows] = await pool.query(
      `SELECT e.event_type, e.dwell_ms, e.post_id, e.created_at,
              e.feed_category, c.slug AS category_slug, pp.tags, pp.user_id AS author_id
       FROM plaza_user_events e
       JOIN plaza_posts pp ON pp.id = e.post_id
       JOIN plaza_categories c ON c.id = pp.category_id
       WHERE e.created_at >= ? ${identityClause}`,
      params,
    );

    const now = Date.now();
    for (const row of eventRows) {
      profile.eventCount += 1;
      profile.seenPostIds.add(row.post_id);
      if (DEEP_SEEN_EVENTS.has(row.event_type)) {
        profile.deepSeenPostIds.add(row.post_id);
      }
      const ageMs = now - Number(row.created_at);
      const decay = decayFactor(ageMs, halfLifeMs);
      const signal =
        eventSignal({ event_type: row.event_type, dwell_ms: row.dwell_ms }) * decay;
      const feedCategory = String(row.feed_category ?? '').trim();
      if (
        feedCategory &&
        (row.event_type === 'impression' || row.event_type === 'click')
      ) {
        bumpWeight(profile.feedCategoryWeights, feedCategory, Math.max(signal, 0.12) * decay);
      }
      if (row.event_type === 'dislike' || row.event_type === 'hide') {
        profile.dislikedPostIds.add(row.post_id);
        profile.dislikedCategorySlugs.add(row.category_slug);
        bumpWeight(profile.categoryWeights, row.category_slug, signal * 0.35);
        continue;
      }
      bumpWeight(profile.categoryWeights, row.category_slug, signal);
      const tags = Array.isArray(row.tags) ? row.tags : JSON.parse(row.tags ?? '[]');
      for (const tag of tags) bumpWeight(profile.tagWeights, String(tag).toLowerCase(), signal * 0.65);
      bumpWeight(profile.authorWeights, row.author_id, signal * 0.8);
      if (row.event_type === 'like') profile.likedPostIds.add(row.post_id);
      if (row.event_type === 'collect') profile.likedPostIds.add(row.post_id);
    }

    if (userId) {
      const [reactions] = await pool.query(
        `SELECT r.type, r.post_id, c.slug AS category_slug, pp.tags, pp.user_id AS author_id, r.created_at
         FROM plaza_reactions r
         JOIN plaza_posts pp ON pp.id = r.post_id AND pp.status = 'published'
         JOIN plaza_categories c ON c.id = pp.category_id
         WHERE r.user_id = ? AND r.created_at >= ?`,
        [userId, since],
      );
      for (const row of reactions) {
        const ageMs = now - Number(row.created_at);
        const decay = decayFactor(ageMs, halfLifeMs);
        const signal = (row.type === 'collect' ? 6.5 : row.type === 'share' ? 3.5 : 5.5) * decay;
        bumpWeight(profile.categoryWeights, row.category_slug, signal);
        const tags = Array.isArray(row.tags) ? row.tags : JSON.parse(row.tags ?? '[]');
        for (const tag of tags) bumpWeight(profile.tagWeights, String(tag).toLowerCase(), signal * 0.65);
        bumpWeight(profile.authorWeights, row.author_id, signal);
        profile.likedPostIds.add(row.post_id);
        profile.seenPostIds.add(row.post_id);
      }

      const [follows] = await pool.query(
        `SELECT followee_id FROM plaza_follows WHERE follower_id = ?`,
        [userId],
      );
      for (const row of follows) {
        profile.followedAuthorIds.add(row.followee_id);
        bumpWeight(profile.authorWeights, row.followee_id, 4.0);
      }

      const [comments] = await pool.query(
        `SELECT c.post_id, cat.slug AS category_slug, pp.tags, pp.user_id AS author_id, c.created_at
         FROM plaza_comments c
         JOIN plaza_posts pp ON pp.id = c.post_id AND pp.status = 'published'
         JOIN plaza_categories cat ON cat.id = pp.category_id
         WHERE c.user_id = ? AND c.status = 'visible' AND c.created_at >= ?`,
        [userId, since],
      );
      for (const row of comments) {
        const ageMs = now - Number(row.created_at);
        const decay = decayFactor(ageMs, halfLifeMs);
        const signal = 4.5 * decay;
        bumpWeight(profile.categoryWeights, row.category_slug, signal);
        const tags = Array.isArray(row.tags) ? row.tags : JSON.parse(row.tags ?? '[]');
        for (const tag of tags) bumpWeight(profile.tagWeights, String(tag).toLowerCase(), signal * 0.65);
        bumpWeight(profile.authorWeights, row.author_id, signal * 0.85);
        profile.seenPostIds.add(row.post_id);
      }
    }

    profile.categoryWeights = normalizeWeights(profile.categoryWeights);
    profile.tagWeights = normalizeWeights(profile.tagWeights);
    profile.authorWeights = normalizeWeights(profile.authorWeights);
    profile.feedCategoryWeights = normalizeWeights(profile.feedCategoryWeights);
    profile.profileVersion =
      1 + profile.eventCount + profile.likedPostIds.size * 3 + profile.dislikedPostIds.size * 5;
    return profile;
  };

  const getProfile = async ({ userId = null, sessionId, halfLifeMs = 7 * 24 * 60 * 60 * 1000 }) => {
    if (!userId && !sessionId) return emptyProfile();
    const key = cacheKey({ userId, sessionId });
    const cached = profileCache.get(key);
    if (cached && cached.expiresAt > Date.now()) {
      return cached.profile;
    }
    const profile = await buildProfileFromSignals({ userId, sessionId, halfLifeMs });
    profileCache.set(key, { profile, expiresAt: Date.now() + PROFILE_CACHE_TTL_MS });
    return profile;
  };

  return {
    recordEvents,
    getProfile,
    invalidateProfileCache,
    internals: {
      emptyProfile,
      decayFactor,
      eventSignal,
      bumpWeight,
      normalizeWeights,
      PLAZA_EVENT_WEIGHTS,
    },
  };
}