Merge pull request #7463 from Countly/newarchitecture

chore: sync next-feature with new-architecture

Author: pnrgenc

CHANGELOG.md

@@ -1,6 +1,38 @@
-## Version 25.03.X
+## Version 25.03.36
+Enterprise fixes:
+- [journey] Workflow fixes
+- [users] UI events table fixes
+
+## Version 25.03.35
+Fixes:
+- [core] Fixes for search bar in standart table component
+
+Enterprise fixes:
+- [journeys] Fixes for journey data updates on incoming data.
+- [surveys] Return error message if invalid widget_id passed on template loading
+- [users] Show content and journey events in user profile
+- [users] Display profile group name in table column
+- [users] When exporting user profiles, replace user name with device id if user name does not exist
+- [users] Use user profile endpoint for exporting data instead of the generic export endpoint
+
+## Version 25.03.34
 Fixes:
 - [core] Fix period calculation
+- [dashboards] Update dialog button color when deleting dashboard/widget
+- [star-rating] Fix rating number when exporting data
+
+Enterprise Fixes:
+- [content] Uniform journey and content block actions
+- [content] Fix overflow and missing translations in content blocks
+- [content] Fix button management when creating fullscreen content blocks
+- [crash_symbolication] Use countlyfs for JavaScript symbolication
+- [funnels] Fix funnel name tooltip content
+- [surveys] Allow surveys to resize and reposition when user rotates devices or adjust browser window
+- [nps] Allow nps to resize and reposition when user rotates devices or adjust browser window
+- [groups] Dealing with invalid values for group permission
+- [geo] Update table row cursor to indicate that it's clickable
+- [users] Change table column min-width to width so it can be resized even smaller
+- [users] Display filtered user count instead of all user count in the table summary
 
 ## Version 25.03.33
 Fixes:

Dockerfile-core

@@ -58,7 +58,7 @@ RUN useradd -r -M -U -d /opt/countly -s /bin/false countly && \
     \
     # npm dependencies
     ./bin/docker/modify.sh && \
-    HOME=/tmp npm install --unsafe-perm && \
+    HOME=/tmp npm ci --unsafe-perm && \
     ./bin/docker/preinstall.sh && \
     \
     # web sdk

Dockerfile-unified

@@ -132,10 +132,14 @@ WORKDIR /opt/countly
 COPY . .
 
 # 3.1.2 copy OpenTelemetry files from build context
+#   otel.js and otelexpress.js both require('./metrics'), so metrics.js
+#   must be copied alongside each entry point.
 RUN if [ -f observability/otel.js ]; then \
         echo "Copying OpenTelemetry files..."; \
         cp observability/otel.js api/utils/otel.js; \
+        cp observability/metrics.js api/utils/metrics.js; \
         cp observability/otelexpress.js frontend/express/otel.js; \
+        cp observability/metrics.js frontend/express/metrics.js; \
     else \
         echo "Observability files not found, skipping..."; \
     fi

Gruntfile.js

@@ -57,6 +57,7 @@ module.exports = function(grunt) {
                     'frontend/express/public/javascripts/utils/Sortable.min.js',
                     'frontend/express/public/javascripts/utils/vue/vuedraggable.umd.min.js',
                     'frontend/express/public/javascripts/utils/lodash.mergeWith.js',
+                    'frontend/express/public/javascripts/utils/cronstrue.min.js',
                     'frontend/express/public/javascripts/utils/element-tiptap.umd.min.js'
                 ],
                 dest: 'frontend/express/public/javascripts/min/countly.utils.concat.js'

api/eventSink/KafkaEventSink.ts

@@ -14,6 +14,28 @@ const EventSinkInterface = require('./EventSinkInterface.ts').default;
 const { transformToKafkaEventFormat } = require('../utils/eventTransformer');
 const Log = require('../utils/log.js');
 
+// OTel metrics — opt-in, zero overhead when OTEL_ENABLED is not set
+const _otelEnabled = /^(1|true|yes)$/i.test(process.env.OTEL_ENABLED || '');
+let _sinkMetrics: { duration: any; eventsTotal: any } | null = null;
+function getSinkMetrics() {
+    if (_sinkMetrics) {
+        return _sinkMetrics;
+    }
+    if (!_otelEnabled) {
+        return null;
+    }
+    try {
+        const { metrics } = require('@opentelemetry/api');
+        const meter = metrics.getMeter('countly-event-sink');
+        _sinkMetrics = {
+            duration: meter.createHistogram('countly_event_sink_duration_seconds', { unit: 's' }),
+            eventsTotal: meter.createCounter('countly_event_sink_events_total'),
+        };
+    }
+    catch (_e) { /* no-op */ }
+    return _sinkMetrics;
+}
+
 /**
  * Logger interface for type safety
  */
@@ -192,6 +214,8 @@ class KafkaEventSink extends EventSinkInterface {
 
             if (result.success) {
                 this.#log.d(`Successfully sent ${result.sent} events to Kafka in ${duration}ms`);
+                getSinkMetrics()?.duration.record(duration / 1000, { sink: 'kafka' });
+                getSinkMetrics()?.eventsTotal.add(result.sent ?? 0, { sink: 'kafka', result: 'success' });
 
                 return this._createResult(true, result.sent ?? 0, 'Events sent to Kafka successfully', {
                     duration,
@@ -205,6 +229,8 @@ class KafkaEventSink extends EventSinkInterface {
         }
         catch (error) {
             const duration = Date.now() - startTime;
+            getSinkMetrics()?.duration.record(duration / 1000, { sink: 'kafka' });
+            getSinkMetrics()?.eventsTotal.add(transformedEvents, { sink: 'kafka', result: 'error' });
             this.#log.e(`Failed to write ${transformedEvents} events to Kafka in ${duration}ms:`);
             this.#log.e('Error writing events to Kafka:', error);
             throw error;

api/eventSource/KafkaEventSource.js

@@ -1,6 +1,33 @@
 const EventSourceInterface = require('./EventSourceInterface');
 const Log = require('../utils/log.js');
 
+// OTel metrics — opt-in, zero overhead when OTEL_ENABLED is not set
+const _otelEnabled = /^(1|true|yes)$/i.test(process.env.OTEL_ENABLED || '');
+let _sourceMetrics = null;
+
+/**
+ * Initialize OTel metrics for event source
+ * @returns {Object|null} OTel metrics object or null if OTel is not enabled
+ * @private
+ */
+function getSourceMetrics() {
+    if (_sourceMetrics) {
+        return _sourceMetrics;
+    }
+    if (!_otelEnabled) {
+        return null;
+    }
+    try {
+        const { metrics } = require('@opentelemetry/api');
+        const meter = metrics.getMeter('countly-event-source');
+        _sourceMetrics = {
+            batchWait: meter.createHistogram('countly_event_source_batch_wait_seconds', { unit: 's' }),
+        };
+    }
+    catch (_e) { /* no-op */ }
+    return _sourceMetrics;
+}
+
 /**
  * Kafka implementation of EventSourceInterface
  * Supports async iteration with auto-acknowledgment and proper at-least-once delivery
@@ -563,6 +590,7 @@ class KafkaEventSource extends EventSourceInterface {
             return checkAndReturnBatch(batch);
         }
         // Wait for next batch from Kafka
+        const waitStart = Date.now();
         await new Promise((resolve) => {
             // Race condition check: batch might have arrived while creating promise
             if (this.#currentBatch) {
@@ -571,6 +599,11 @@ class KafkaEventSource extends EventSourceInterface {
             }
             this.#batchAvailable = resolve;
         });
+        const waitDuration = (Date.now() - waitStart) / 1000;
+        getSourceMetrics()?.batchWait.record(waitDuration, {
+            group_id: this.#effectiveGroupId || this.#name,
+            topic: this.#kafkaOptions?.topics?.[0] || 'unknown'
+        });
 
         // Return the batch that arrived
         if (this.#currentBatch) {

api/ingestor/requestProcessor.ts

@@ -9,7 +9,6 @@ import { createRequire } from 'module';
 
 import usage from './usage.js';
 import common from '../utils/common.js';
-import url from 'url';
 import logModule from '../utils/log.js';
 import crypto from 'crypto';
 import { ignorePossibleDevices, checksumSaltVerification, validateRedirect } from '../utils/requestProcessorCommon.js';
@@ -378,7 +377,7 @@ interface RequestParams {
     /** Href */
     href?: string;
     /** URL parts */
-    urlParts?: ReturnType<typeof url.parse>;
+    urlParts?: { pathname: string; path: string };
     /** Path segments */
     paths?: string[];
     /** API path */
@@ -1246,14 +1245,15 @@ const processRequest = (params: RequestParams): boolean | void => {
     }
 
     params.tt = Date.now().valueOf();
-    const urlParts = url.parse(params.req.url, true);
-    const queryString = urlParts.query;
-    const paths = urlParts.pathname.split('/');
+    // base URL is required by WHATWG URL API for relative paths, only pathname and query are used
+    const parsedUrl = new URL(params.req.url, 'http://localhost');
+    const queryString = Object.fromEntries(parsedUrl.searchParams);
+    const paths = parsedUrl.pathname.split('/');
 
-    params.href = urlParts.href;
+    params.href = parsedUrl.pathname + parsedUrl.search;
     params.qstring = params.qstring || {};
     params.res = params.res || {} as ServerResponse;
-    params.urlParts = urlParts;
+    params.urlParts = { pathname: parsedUrl.pathname, path: parsedUrl.pathname + parsedUrl.search };
     params.paths = paths;
 
     params.req.headers = params.req.headers || {};

api/jobs/mutationManagerJob.ts

@@ -26,11 +26,12 @@ interface JobConfig {
 
 interface MutationTask extends Document {
     _id: ObjectId | string;
-    type: 'delete' | 'update';
+    type: 'delete' | 'update' | 'native_ch';
     db: string;
     collection: string;
     query: Record<string, unknown>;
     update?: Record<string, unknown>;
+    native_sql?: string;
     running: boolean;
     status: string;
     hb?: number;
@@ -272,7 +273,7 @@ class MutationManagerJob extends Job {
      */
     async processTask(task: MutationTask, summary: SummaryEntry[], jobConfig: JobConfig = jobConfigState || DEFAULT_JOB_CONFIG): Promise<void> {
         const type = task.type;
-        if (type !== 'delete' && type !== 'update') {
+        if (type !== 'delete' && type !== 'update' && type !== 'native_ch') {
             await common.db.collection('mutation_manager').updateOne(
                 { _id: task._id },
                 {
@@ -288,7 +289,7 @@ class MutationManagerJob extends Job {
         const clickhouseEnabled = mutationManager.isClickhouseEnabled();
         const hasClickhouseDelete = clickhouseEnabled && !!(clickHouseRunner && clickHouseRunner.deleteGranularDataByQuery);
         const hasClickhouseUpdate = clickhouseEnabled && !!(clickHouseRunner && clickHouseRunner.updateGranularDataByQuery);
-        const hasClickhouse = (type === 'update' ? hasClickhouseUpdate : hasClickhouseDelete);
+        const hasClickhouse = type === 'native_ch' ? clickhouseEnabled : (type === 'update' ? hasClickhouseUpdate : hasClickhouseDelete);
 
         if (!mongoDb && !hasClickhouse) {
             const reason = `mongo_db_unavailable:${task.db || 'missing'}`;
@@ -317,7 +318,11 @@ class MutationManagerJob extends Job {
         }
 
         let mongoOk = true;
-        if (mongoDb) {
+        if (type === 'native_ch') {
+            // Native CH mutations skip MongoDB entirely
+            log.d('Native CH mutation - skipping MongoDB', { taskId: task._id });
+        }
+        else if (mongoDb) {
             if (type === 'update') {
                 mongoOk = await this.updateMongo(task, mongoDb);
             }
@@ -330,7 +335,10 @@ class MutationManagerJob extends Job {
         }
 
         let chScheduledOk = true;
-        if (type === 'update' && hasClickhouseUpdate) {
+        if (type === 'native_ch' && clickhouseEnabled) {
+            chScheduledOk = await this.executeNativeClickhouse(task);
+        }
+        else if (type === 'update' && hasClickhouseUpdate) {
             chScheduledOk = await this.updateClickhouse(task);
         }
         else if (type === 'delete' && hasClickhouseDelete) {
@@ -422,8 +430,10 @@ class MutationManagerJob extends Job {
         for (const task of awaiting) {
             try {
                 if (chHealth && typeof chHealth.getMutationStatus === 'function') {
-                    // In cluster mode, mutations target _local tables, so validation must check _local
-                    const validationTable = isClusterMode ? task.collection + '_local' : task.collection;
+                    // In cluster mode, mutations target _local tables, so validation must check _local.
+                    // native_ch tasks may already have _local in collection name — avoid doubling.
+                    const needsLocalSuffix = isClusterMode && !task.collection.endsWith('_local');
+                    const validationTable = needsLocalSuffix ? task.collection + '_local' : task.collection;
                     const status = await chHealth.getMutationStatus({ validation_command_id: task.validation_command_id, table: validationTable, database: task.db });
                     if (status && status.is_done) {
                         await common.db.collection('mutation_manager').updateOne(
@@ -677,6 +687,95 @@ class MutationManagerJob extends Job {
         }
     }
 
+    /**
+     * Build validated ClickHouse mutation SQL with an embedded command-id for tracking.
+     * - Strips trailing semicolon
+     * - Validates ALTER TABLE ... DELETE/UPDATE ... WHERE ... shape
+     * - Injects tautological AND before any SETTINGS clause
+     * @returns Final SQL string, or null if the shape is invalid
+     */
+    private buildValidatedNativeClickhouseSql(baseSql: string, commandId: string): string | null {
+        if (!baseSql || typeof baseSql !== 'string') {
+            return null;
+        }
+        let sql = baseSql.trim();
+        if (sql.endsWith(';')) {
+            sql = sql.slice(0, -1).trimEnd();
+        }
+        const upper = sql.toUpperCase();
+        if (!upper.startsWith('ALTER TABLE ')) {
+            return null;
+        }
+        if (!/\b(DELETE|UPDATE)\b/.test(upper)) {
+            return null;
+        }
+        if (!/\bWHERE\b/.test(upper)) {
+            return null;
+        }
+        // Find SETTINGS clause (if any) — inject command-id BEFORE it
+        const settingsMatch = upper.match(/\bSETTINGS\b/);
+        const settingsIdx = settingsMatch?.index ?? -1;
+        const injection = ` AND '${commandId}' = '${commandId}'`;
+        if (settingsIdx !== -1) {
+            return sql.slice(0, settingsIdx) + injection + sql.slice(settingsIdx);
+        }
+        return sql + injection;
+    }
+
+    /**
+     * Executes a native ClickHouse SQL mutation directly.
+     * Used for complex mutations (e.g., deduplication) that cannot be expressed as Mongo-style queries.
+     * Embeds validation_command_id for tracking via system.mutations.
+     * @param task - The mutation task with native_sql field
+     */
+    async executeNativeClickhouse(task: MutationTask): Promise<boolean> {
+        if (!task.native_sql || typeof task.native_sql !== 'string') {
+            log.e('Skipping native CH mutation (empty sql)', { taskId: task._id });
+            await this.markFailedOrRetry(task, 'empty_native_sql');
+            return false;
+        }
+
+        if (!common.clickhouseQueryService) {
+            log.e('ClickHouse query service not available for native mutation', { taskId: task._id });
+            await this.markFailedOrRetry(task, 'ch_query_service_unavailable');
+            return false;
+        }
+
+        try {
+            const retryIndex = Number(task.fail_count || 0);
+            const commandId = `nm_${String(task._id)}_${retryIndex}`;
+
+            const sql = this.buildValidatedNativeClickhouseSql(task.native_sql, commandId);
+            if (!sql) {
+                log.e('Skipping native CH mutation (invalid SQL shape)', {
+                    taskId: task._id,
+                    native_sql: task.native_sql
+                });
+                await this.markFailedOrRetry(task, 'invalid_native_sql_shape');
+                return false;
+            }
+
+            // Persist command_id BEFORE executing mutation (crash safety: if we crash
+            // between execution and this update, validation can still find the command_id)
+            await common.db.collection('mutation_manager').updateOne(
+                { _id: task._id },
+                { $set: { validation_command_id: commandId } }
+            );
+
+            await common.clickhouseQueryService.executeMutation({ query: sql });
+            log.d('Native CH mutation scheduled', { taskId: task._id, commandId });
+            return true;
+        }
+        catch (err) {
+            log.e('Native CH mutation failed', {
+                taskId: task._id,
+                error: (err as Error)?.message || String(err)
+            });
+            await this.markFailedOrRetry(task, 'native_ch_error: ' + ((err as Error)?.message || err + ''));
+            return false;
+        }
+    }
+
     /**
      * Marks a task as failed or schedules it for a retry based on the number of previous failures.
      * @param task - The task object to update.