Function pushImport

packages/node-runtime/src/services/push-importer.ts:431–518

(
  dbManager: DatabaseManager,
  sessionId: string,
  payload: PushImportPayload
)

const SAFE_SESSION_ID_RE = /^[a-zA-Z0-9_@-][a-zA-Z0-9_@.-]*$/

export async function pushImport(
  dbManager: DatabaseManager,
  sessionId: string,
  payload: PushImportPayload
): Promise<PushImportOutcome> {
  if (!SAFE_SESSION_ID_RE.test(sessionId) || sessionId.includes('..')) {
    return { ok: false, reason: 'invalid_payload', message: 'sessionId contains invalid characters' }
  }

  if (importInProgress.has(sessionId)) {
    return {
      ok: false,
      reason: 'import_in_progress',
      message: 'Another import is already in progress for this session',
    }
  }

  const dbPath = dbManager.getDbPath(sessionId)
  const isNew = !fs.existsSync(dbPath)

  const validationError = validatePayload(payload, isNew)
  if (validationError) {
    return { ok: false, reason: 'invalid_payload', message: validationError }
  }

  importInProgress.add(sessionId)
  try {
    if (isNew) {
      const db = dbManager.openRawSessionDatabase(sessionId, { create: true })
      try {
        const { writtenCount, duplicateCount, membersAdded } = fullImport(
          db,
          payload.meta!,
          payload.members ?? [],
          payload.messages!
        )
        const session = queryStats(db)
        dbManager.raiseCurrentChatDbCompatibilityGate()
        return {
          ok: true,
          result: {
            sessionId,
            created: true,
            batch: { receivedCount: payload.messages!.length, writtenCount, duplicateCount },
            session,
            updates: { metaUpdated: true, membersAdded, membersUpdated: 0 },
          },
        }
      } finally {
        db.close()
      }
    }

    const db = dbManager.openRawSessionDatabase(sessionId, { readonly: false })
    try {
      const { writtenCount, duplicateCount, metaUpdated, membersAdded, membersUpdated } = incrementalImport(db, payload)
      const session = queryStats(db)
      dbManager.raiseCurrentChatDbCompatibilityGate()