memind/health-page-data-observation-store.mjs

import { createUserDataSpaceService } from './user-data-space-service.mjs';
import { ensureHealthPageDataForUser } from './health-page-data-bootstrap.mjs';
import { HEALTH_OBSERVATIONS_DATASET } from './health-page-data-schema.mjs';
import { mapObservationToPgPayload, mapPgObservationRow } from './health-page-data-map.mjs';
import { observationDedupKey } from './health-observation-validate.mjs';

const DATASET_NAME = HEALTH_OBSERVATIONS_DATASET.name;

export function createPageDataHealthObservationStore({
  resolveWorkspaceRoot,
  logger = console,
} = {}) {
  if (typeof resolveWorkspaceRoot !== 'function') {
    throw new Error('createPageDataHealthObservationStore requires resolveWorkspaceRoot');
  }

  async function getUserDataSpace(userId) {
    const workspaceRoot = await resolveWorkspaceRoot(userId);
    if (!workspaceRoot) return null;
    const service = createUserDataSpaceService({ workspaceRoot, userId: String(userId) });
    await ensureHealthPageDataForUser(service, userId);
    return service;
  }

  async function list(userId, { limit = 50 } = {}) {
    const service = await getUserDataSpace(userId);
    if (!service) return [];
    try {
      const result = await service.readDatasetRows(DATASET_NAME, {
        limit: Math.min(Math.max(limit, 1), 500),
        orderBy: 'observed_at',
        orderDir: 'desc',
      });
      return (result.rows ?? []).map((row) => mapPgObservationRow(row, userId));
    } catch (error) {
      if (error?.code === 'dataset_not_found') return [];
      throw error;
    }
  }

  async function findByDedupKey(userId, dedupKey) {
    if (!dedupKey) return null;
    const parts = String(dedupKey).split('|');
    if (parts.length < 3) return null;
    const [metricType, observedAt, sourceRef] = parts;
    const rows = await list(userId, { limit: 500 });
    return (
      rows.find(
        (row) =>
          row.metricType === metricType
          && String(row.observedAt) === String(observedAt)
          && row.sourceRef === sourceRef,
      ) ?? null
    );
  }

  async function insert(userId, observation) {
    if (!observation?.confirmed) {
      const error = Object.assign(new Error('健康观测必须确认后才能保存'), {
        code: 'health_unconfirmed',
      });
      throw error;
    }

    const dedup = observation.dedupKey
      ?? observationDedupKey({
        metricType: observation.metricType,
        observedAt: observation.observedAt,
        sourceRef: observation.sourceRef ?? null,
      });
    if (dedup) {
      const existing = await findByDedupKey(userId, dedup);
      if (existing) return existing;
    }

    const service = await getUserDataSpace(userId);
    if (!service) {
      const error = Object.assign(new Error('用户健康数据空间不可用'), { code: 'health_storage_unavailable' });
      throw error;
    }

    try {
      const { row } = await service.insertDatasetRow(DATASET_NAME, mapObservationToPgPayload(observation));
      return mapPgObservationRow(row, userId);
    } catch (error) {
      if (dedup && /duplicate key|unique constraint/i.test(String(error?.message ?? ''))) {
        const existing = await findByDedupKey(userId, dedup);
        if (existing) return existing;
      }
      logger.warn?.('Page Data health observation insert failed:', error);
      throw error;
    }
  }

  return {
    backend: 'page_data',
    list,
    findByDedupKey,
    insert,
  };
}