apps/start/src/modals/index.tsx (1)

8-8: Consider lazy‐loading AddImport to keep initial bundle slim

AddImport brings TRPC/React Query; load it on demand.

Apply:

+import { lazy } from 'react';
-import AddImport from './add-import';
+const AddImport = lazy(() => import('./add-import'));

Ensure ModalProvider or the modal surface renders within a .

Also applies to: 41-41

packages/importer/src/base-provider.ts (1)

7-8: Make core hooks async‐friendly and identifiers immutable

Allow providers to perform async transforms/validation; mark identifiers readonly.

 export abstract class BaseImportProvider<
 TRawEvent extends BaseRawEvent = BaseRawEvent,
 > {
- abstract provider: string;
- abstract version: string;
+ abstract readonly provider: string;
+ abstract readonly version: string;
+
+ // Utility type to allow sync or async implementations
+ export type MaybePromise<T> = T | Promise<T>;
 /**
 * Stream-read and parse source (file/API) → yields raw events
 * This should be implemented as an async generator to handle large files efficiently
 */
 abstract parseSource(): AsyncGenerator<TRawEvent, void, unknown>;
 /**
 * Convert provider format → IClickhouseEvent
 */
- abstract transformEvent(rawEvent: TRawEvent): IClickhouseEvent;
+ abstract transformEvent(rawEvent: TRawEvent): MaybePromise<IClickhouseEvent>;
 /**
 * Validate raw event structure
 */
- abstract validate(rawEvent: TRawEvent): boolean;
+ abstract validate(rawEvent: TRawEvent): MaybePromise<boolean>;

Also applies to: 19-24

packages/importer/src/utils/referrer-parser.ts (3)

5-35: Type the map and normalize hosts; reduce duplicates

Strong types + normalization avoid listing www/m. variants twice and improve coverage.

-import { stripTrailingSlash } from '@openpanel/common';
+import { stripTrailingSlash } from '@openpanel/common';
+type ReferrerType = 'search' | 'social' | 'referral';
+type ReferrerEntry = { type: ReferrerType; name: string };
+const normalizeHost = (h: string) => h.replace(/^(www\.|m\.)/i, '').toLowerCase();
-const referrers: Record<string, { type: string; name: string }> = {
+const referrers: Record<string, ReferrerEntry> = {
 // Search engines
- 'google.com': { type: 'search', name: 'Google' },
- 'www.google.com': { type: 'search', name: 'Google' },
+ 'google.com': { type: 'search', name: 'Google' },
 'bing.com': { type: 'search', name: 'Bing' },
- 'www.bing.com': { type: 'search', name: 'Bing' },
 'yahoo.com': { type: 'search', name: 'Yahoo!' },
- 'www.yahoo.com': { type: 'search', name: 'Yahoo!' },
 'duckduckgo.com': { type: 'search', name: 'DuckDuckGo' },
- 'www.duckduckgo.com': { type: 'search', name: 'DuckDuckGo' },
 'baidu.com': { type: 'search', name: 'Baidu' },
- 'www.baidu.com': { type: 'search', name: 'Baidu' },
 // Social media
- 'facebook.com': { type: 'social', name: 'Facebook' },
- 'www.facebook.com': { type: 'social', name: 'Facebook' },
+ 'facebook.com': { type: 'social', name: 'Facebook' },
 'twitter.com': { type: 'social', name: 'Twitter' },
- 'www.twitter.com': { type: 'social', name: 'Twitter' },
 'x.com': { type: 'social', name: 'X' },
- 'www.x.com': { type: 'social', name: 'X' },
 'linkedin.com': { type: 'social', name: 'LinkedIn' },
- 'www.linkedin.com': { type: 'social', name: 'LinkedIn' },
 'instagram.com': { type: 'social', name: 'Instagram' },
- 'www.instagram.com': { type: 'social', name: 'Instagram' },
 'tiktok.com': { type: 'social', name: 'TikTok' },
- 'www.tiktok.com': { type: 'social', name: 'TikTok' },
 'youtube.com': { type: 'social', name: 'YouTube' },
- 'www.youtube.com': { type: 'social', name: 'YouTube' },
 'reddit.com': { type: 'social', name: 'Reddit' },
- 'www.reddit.com': { type: 'social', name: 'Reddit' },
+ 't.co': { type: 'social', name: 'Twitter' },
+ 'lnkd.in': { type: 'social', name: 'LinkedIn' },
 };
 
 export function parseReferrer(url: string | undefined) {
- const hostname = getHostname(url);
- const match = referrers[hostname] ?? referrers[hostname.replace('www.', '')];
+ const hostname = normalizeHost(getHostname(url));
+ const match = referrers[hostname];

Also applies to: 49-52

---

49-57: Guard output URL to http(s) only

Avoid echoing non-web schemes (e.g., javascript:) back to callers.

 export function parseReferrer(url: string | undefined) {
 const hostname = getHostname(url);
 const match = referrers[hostname] ?? referrers[hostname.replace('www.', '')];
- return {
+ let safeUrl = '';
+ try {
+ if (url) {
+ const u = new URL(url);
+ if (u.protocol === 'http:' || u.protocol === 'https:') {
+ safeUrl = stripTrailingSlash(url);
+ }
+ }
+ } catch {
+ // ignore
+ }
+
+ return {
 name: match?.name ?? '',
 type: match?.type ?? 'referral',
- url: stripTrailingSlash(url ?? ''),
+ url: safeUrl,
 };
 }

---

60-91: Minor: normalize source before lookup and keep types consistent

Lower-case once; reuse normalizeHost for direct key match; return typed result.

-export function getReferrerWithQuery(
- query: Record<string, string> | undefined,
-) {
+export function getReferrerWithQuery(
+ query: Record<string, string> | undefined,
+): { name: string; type: ReferrerType; url: string } | null {
 if (!query) {
 return null;
 }
- const source = query.utm_source ?? query.ref ?? query.utm_referrer ?? '';
+ const source = (query.utm_source ?? query.ref ?? query.utm_referrer ?? '').trim();
 if (source === '') {
 return null;
 }
- const match =
- Object.values(referrers).find(
- (referrer) => referrer.name.toLowerCase() === source.toLowerCase(),
- ) || referrers[source];
+ const lower = source.toLowerCase();
+ const match =
+ Object.values(referrers).find((r) => r.name.toLowerCase() === lower) ||
+ referrers[normalizeHost(lower)];
 if (match) {
 return {
 name: match.name,
 type: match.type,
 url: '',
 };
 }

packages/db/prisma/migrations/20251018214030_fix/migration.sql (1)

1-13: Consider more descriptive migration naming.

The migration name "fix" is not descriptive. Consider renaming to something like "remove_redundant_import_columns" or "migrate_import_config_to_json" to better indicate the purpose of this schema change.

packages/db/code-migrations/5-add-imports-table.sql (1)

32-32: Consider using an Enum for import_status.

The import_status field uses LowCardinality(String) with a default of 'pending'. For better type safety and query performance, consider using an Enum type instead:

`import_status` LowCardinality(Enum8('pending' = 1, 'processing' = 2, 'completed' = 3, 'failed' = 4)) DEFAULT 'pending'

This ensures only valid status values can be stored and provides better compression.

packages/db/prisma/schema.prisma (1)

470-486: Consider using an enum for the status field.

The status field is defined as a String with a comment indicating it should be one of 'pending' | 'processing' | 'completed' | 'failed'. For better type safety and database constraints, define a Prisma enum:

enum ImportStatus {
 pending
 processing
 completed
 failed
}
model Import {
 // ...
 status ImportStatus // instead of String
 // ...
}

This provides compile-time type checking and database-level validation.

packages/trpc/src/root.ts (1)

7-7: Router key named "import" (reserved keyword) — consider renaming for DX.

Valid in JS/TS, but can confuse tooling/usages (e.g., appRouter.import vs dynamic import()). Consider renaming to imports/importer for clarity.

Also applies to: 44-44

packages/queue/src/queues.ts (1)

163-176: Harden import jobs: add attempts/backoff/timeout; optionally expose QueueEvents.

Imports call external providers; transient failures are expected. Provide sensible defaults and metrics hook.

Apply:

 export const importQueue = new Queue<ImportQueuePayload>('import', {
 connection: getRedisQueue(),
 defaultJobOptions: {
 removeOnComplete: 10,
- removeOnFail: 50,
+ removeOnFail: 50,
+ attempts: 3,
+ backoff: { type: 'exponential', delay: 30_000 },
+ timeout: 1000 * 60 * 30, // 30 min; tune as needed
 },
 });
+
+// Optional: enable events for monitoring dashboards/Bull Board progress
+export const importQueueEvents = new QueueEvents('import', {
+ connection: getRedisQueue(),
+});

apps/worker/src/boot-workers.ts (3)

138-142: Also drain import queue before shutdown to avoid cutting imports mid-run.

- await waitForQueueToEmpty(cronQueue);
+ await waitForQueueToEmpty(cronQueue);
+ await waitForQueueToEmpty(importQueue, 10 * 60_000); // 10 min cap; tune as needed

---

155-167: Dev path skips graceful shutdown — can leave jobs in inconsistent state.

Unify behavior and call the same exit handler in dev.

- process.on(evt, (code) => {
- if (process.env.NODE_ENV === 'production') {
- exitHandler(evt, code);
- } else {
- logger.info('Shutting down for development', {
- event: evt,
- code,
- });
- process.exit(0);
- }
- });
+ process.on(evt, (code) => {
+ exitHandler(evt, code ?? 0);
+ });

---

4-13: Remove unused imports and dead/commented code for clarity.

eventsGroupQueue, GroupWorker, eventsJob, incomingEventPure are no longer used; commented worker block can be dropped.

Also applies to: 18-23, 33-48

packages/trpc/src/routers/import.ts (3)

121-131: Guard job removal on delete with error handling (active/locked jobs can throw).

 const job = await importQueue.getJob(importRecord.jobId);
- if (job) {
- await job.remove();
- }
+ if (job) {
+ try {
+ await job.remove();
+ } catch (e) {
+ // best-effort removal; proceed with deletion
+ }
+ }

---

165-171: Retry should set stable jobId and attempts/backoff like create.

- const job = await importQueue.add('import', {
- type: 'import',
- payload: {
- importId: importRecord.id,
- },
- });
+ const job = await importQueue.add(
+ 'import',
+ { type: 'import', payload: { importId: importRecord.id } },
+ { jobId: importRecord.id, attempts: 3, backoff: { type: 'exponential', delay: 30_000 } },
+ );

---

3-6: Unused type import (IImportConfig).

IImportConfig is not referenced. Remove to satisfy lint.

packages/importer/src/index.ts (1)

1-14: Public API surface looks good; minor polish

Consider also re-exporting provider config/types (e.g., IUmamiImportConfig) from providers to keep consumers on a single import path. Optional.

IMPORT_REVIEW.md (1)

320-337: Move this review doc to /docs or PR notes to avoid shipping in package tarballs

Keeps repo clean and avoids confusing consumers.

apps/worker/src/jobs/import.ts (4)

1-11: Avoid deep imports; prefer package barrel exports

Importing services via '@openpanel/db/src/services/import.service' couples to internal paths. Re-export these from '@openpanel/db' and import from there to reduce breakage.

---

75-81: Progress totalBatches fluctuates; compute once or report pct

As totalEvents grows, totalBatches changes mid-run. Either omit totalBatches here or compute a percentage for UI.

- totalBatches: Math.ceil(totalEvents / BATCH_SIZE),
+ progressPct: totalEvents ? Math.round((processedEvents / totalEvents) * 100) : 0,

---

17-17: Make BATCH_SIZE configurable

Read from config/env to tune per deployment.

-const BATCH_SIZE = 5000;
+const BATCH_SIZE = Number(process.env.IMPORT_BATCH_SIZE ?? 5000);

---

200-209: Validate provider-specific config with a schema before instantiation

Avoid passing untyped Prisma JSON into providers. Parse with provider schema (zod) and fail early.

 function createProvider(
 record: Prisma.ImportGetPayload<{ include: { project: true } }>,
 ) {
 switch (record.config.provider) {
 case 'umami':
- return new UmamiProvider(record.projectId, record.config);
+ // Example: zUmamiImportConfig.safeParse(record.config)
+ return new UmamiProvider(
+ record.projectId,
+ (record.config as any) /* TODO: parse/validate */,
+ );
 default:
 throw new Error(`Unknown provider: ${record.config.provider}`);
 }
 }

I can wire in zod parsing if you expose the provider config schemas from @openpanel/importer.

packages/db/prisma/migrations/20251018205153_add_import_table/migration.sql (2)

9-16: Constrain status values

Add a CHECK constraint (or enum) to prevent invalid states.

 "status" TEXT NOT NULL,
@@
 ) ;
+
+-- Restrict status values
+ALTER TABLE "public"."imports"
+ ADD CONSTRAINT "imports_status_check"
+ CHECK ("status" IN ('pending','processing','completed','failed'));

---

18-22: Add indexes for common queries

List/filters will hit projectId/status/createdAt frequently.

 -- AddForeignKey
 ALTER TABLE "public"."imports" ADD CONSTRAINT "imports_projectId_fkey" FOREIGN KEY ("projectId") REFERENCES "public"."projects"("id") ON DELETE CASCADE ON UPDATE CASCADE;
+
+-- Indexes for performance
+CREATE INDEX IF NOT EXISTS "idx_imports_project_createdAt"
+ ON "public"."imports" ("projectId", "createdAt" DESC);
+CREATE INDEX IF NOT EXISTS "idx_imports_status_createdAt"
+ ON "public"."imports" ("status", "createdAt" DESC);

apps/start/src/routes/_app.$organizationId.$projectId_.settings._tabs.imports.tsx (5)

5-10: Remove unused icon import

CheckCircle is unused.

-import {
- CheckCircle,
- CheckCircleIcon,
- Download,
- Loader2Icon,
- XCircleIcon,
-} from 'lucide-react';
+import { CheckCircleIcon, Download, Loader2Icon, XCircleIcon } from 'lucide-react';

---

83-104: Add onError to mutations for user feedback

Surface errors with toasts; also consider logging.

 const deleteImport = useMutation(
 trpc.import.delete.mutationOptions({
 onSuccess: () => {
 toast.success('Import deleted', {
 description: 'The import has been successfully deleted.',
 });
 queryClient.invalidateQueries(trpc.import.list.pathFilter());
 },
+ onError: (err) => {
+ toast.error('Delete failed', { description: String(err) });
+ },
 }),
 );
@@
 const retryImport = useMutation(
 trpc.import.retry.mutationOptions({
 onSuccess: () => {
 toast.success('Import retried', {
 description: 'The import has been queued for processing again.',
 });
 queryClient.invalidateQueries(trpc.import.list.pathFilter());
 },
+ onError: (err) => {
+ toast.error('Retry failed', { description: String(err) });
+ },
 }),
 );

---

244-261: Disable action buttons while pending to avoid double submits

Tie disabled state to mutation pending flags.

- {imp.status === 'failed' && (
+ {imp.status === 'failed' && (
 <Button
 variant="outline"
 size="sm"
- onClick={() => retryImport.mutate({ id: imp.id })}
+ disabled={retryImport.isPending}
+ onClick={() => retryImport.mutate({ id: imp.id })}
 >
 Retry
 </Button>
 )}
 <Button
 variant="ghost"
 size="sm"
- onClick={() => deleteImport.mutate({ id: imp.id })}
+ disabled={deleteImport.isPending}
+ onClick={() => deleteImport.mutate({ id: imp.id })}
 >
 Delete
 </Button>

---

41-66: Consider serving logos locally or via asset proxy

External Brandfetch URLs can break and may violate CSP. If feasible, mirror assets or use a proxy.

---

141-161: Consider extending IntegrationCardLogoImage to accept a customizable alt prop for improved accessibility

The alt attribute is currently hardcoded to the generic "Integration Logo" at line 66 of apps/start/src/components/integrations/integration-card.tsx. Extending the component to accept an optional alt prop and passing provider.name would provide more descriptive alternative text for each integration, improving accessibility and SEO.

packages/validation/src/index.ts (1)

444-445: Local file support vs schema

fileUrl is z.string().url(), but the Umami provider supports local files. Either (a) allow file:/// or absolute paths, or (b) drop local-path support in the provider. Today this schema blocks local imports.

- fileUrl: z.string().url(),
+ fileUrl: z.string().refine(
+ (v) => /^https?:\/\//i.test(v) || /^file:\/\//i.test(v) || v.startsWith('/'),
+ 'Must be an http(s) URL, file:// URL, or absolute path',
+ ),

packages/importer/src/providers/umami.ts (2)

178-195: Byte counting via body.on('data') competes with pipeline

Attach a lightweight Transform to count bytes between body and decompress to preserve backpressure and avoid dual consumers.

- if (maxBytes) {
- body.on('data', (chunk: Buffer) => {
- seenBytes += chunk.length;
- if (seenBytes > maxBytes) {
- controller.abort();
- body.destroy(new Error(`Stream exceeded size limit (${seenBytes} > ${maxBytes})`));
- }
- });
- }
+ const byteCounter = new (await import('node:stream')).Transform({
+ transform(chunk, _enc, cb) {
+ seenBytes += chunk.length;
+ if (maxBytes && seenBytes > maxBytes) {
+ cb(new Error(`Stream exceeded size limit (${seenBytes} > ${maxBytes})`));
+ return;
+ }
+ cb(null, chunk);
+ },
+ });
@@
- if (decompress) {
- await pipeline(body, decompress, parser, { signal: controller.signal });
- } else {
- await pipeline(body, parser, { signal: controller.signal });
- }
+ if (decompress) {
+ await pipeline(body, byteCounter, decompress, parser, { signal: controller.signal });
+ } else {
+ await pipeline(body, byteCounter, parser, { signal: controller.signal });
+ }

Also applies to: 211-227

---

91-99: Optional: support local Brotli (.br) like remote path

Mirror remote handling and support .br locally if needed.

- const isGzipped = filePath.endsWith('.gz');
+ const isGzipped = filePath.endsWith('.gz');
+ const isBrotli = filePath.endsWith('.br');
@@
- if (isGzipped) {
+ if (isGzipped) {
 await pipeline(fileStream, createGunzip(), parser);
- } else {
+ } else if (isBrotli) {
+ await pipeline(fileStream, createBrotliDecompress(), parser);
+ } else {
 await pipeline(fileStream, parser);
 }

packages/db/src/services/import.service.ts (3)

78-92: Consider using CTEs or subqueries for better maintainability.

The argMin/argMax tuple approach works but is fragile—the 17-element tuple order must match the destructuring on lines 147-173. If fields are added, removed, or reordered, this becomes error-prone and difficult to debug.

A clearer alternative would be to use CTEs with ROW_NUMBER() or separate queries for first/last events, making field selection explicit:

WITH ranked_events AS (
 SELECT *,
 ROW_NUMBER() OVER (PARTITION BY device_id, session_id, project_id, profile_id ORDER BY created_at ASC) as first_rank,
 ROW_NUMBER() OVER (PARTITION BY device_id, session_id, project_id, profile_id ORDER BY created_at DESC) as last_rank
 FROM ${TABLE_NAMES.events_imports}
 WHERE ${rangeWhere} AND name NOT IN ('session_start', 'session_end')
)
SELECT ...
FROM ranked_events
WHERE first_rank = 1 OR last_rank = 1

---

26-525: Consider adding explicit error handling and logging.

None of the functions include try-catch blocks or error enrichment. While throwing errors to the caller is acceptable, adding contextual error handling would improve debuggability and allow for better error classification (transient vs. permanent failures).

Consider wrapping critical operations:

export async function insertImportBatch(
 events: IClickhouseEvent[],
 importId: string,
): Promise<ImportStageResult> {
 try {
 if (events.length === 0) {
 return { importId, totalEvents: 0, insertedEvents: 0 };
 }
 
 const eventsWithMetadata = events.map((event) => ({
 ...event,
 import_id: importId,
 import_status: 'pending',
 imported_at_meta: new Date(),
 }));
 await ch.insert({
 table: TABLE_NAMES.events_imports,
 values: eventsWithMetadata,
 format: 'JSONEachRow',
 });
 return {
 importId,
 totalEvents: events.length,
 insertedEvents: events.length,
 };
 } catch (error) {
 // Add context and rethrow or log
 throw new Error(`Failed to insert import batch for ${importId}: ${error instanceof Error ? error.message : String(error)}`);
 }
}

This pattern would help with debugging import failures in production.

---

278-356: Consider batch migration for very large imports.

The INSERT INTO ... SELECT approach is efficient for most cases, but for imports with millions of events, it might:

• Block for extended periods
• Consume significant memory
• Provide no progress feedback during the migration

For very large imports, consider implementing batch migration:

export async function migrateImportToProduction(
 importId: string,
 from?: string,
 to?: string,
 batchSize = 100000, // configurable batch size
): Promise<void> {
 let offset = 0;
 let migrated = 0;
 
 while (true) {
 const migrationQuery = `
 INSERT INTO ${TABLE_NAMES.events} (...)
 SELECT ...
 FROM ${TABLE_NAMES.events_imports}
 WHERE import_id = {importId:String}
 ${from ? 'AND created_at >= {from:String}' : ''}
 ${to ? 'AND created_at < {to:String}' : ''}
 ORDER BY created_at ASC
 LIMIT {batchSize:UInt32} OFFSET {offset:UInt32}
 `;
 
 const result = await ch.command({
 query: migrationQuery,
 query_params: { importId, from, to, batchSize, offset },
 });
 
 // If no rows affected, we're done
 if (result.query_id && migrated === 0) break;
 
 offset += batchSize;
 // Could call updateImportProgress here
 }
}

This is optional and depends on expected import sizes.

Configuration used: CodeRabbit UI

Review profile: CHILL

Plan: Pro