feat(store+cypher): channel dedup, count(DISTINCT), SQL injection fix

Channel deduplication:
- Added UNIQUE index on channels(project, channel_name, direction,
  file_path, function_name) to prevent duplicate rows at insert time
- Changed INSERT to INSERT OR IGNORE
- Added DISTINCT to all channel SELECT queries
- Fixed SQL injection in channel DELETE (was snprintf, now parameterized)

Cypher count(DISTINCT ...):
- Parser now accepts DISTINCT keyword inside aggregate functions:
  count(DISTINCT n.name), count(DISTINCT n.file_path), etc.
- Added distinct_arg flag to cbm_return_item_t
- Executor tracks seen values per-column and only increments count
  for unique values when distinct_arg is set
- Proper cleanup of distinct_seen arrays in both WITH and RETURN paths

Enables queries like:
  MATCH (caller)-[e]->(n) WHERE e.type = 'CALLS'
  RETURN count(DISTINCT n.name) as unique_callees

Author: Your Name

src/cypher/cypher.c

@@ -1052,6 +1052,10 @@ static int parse_return_or_with(parser_t *p, cbm_return_clause_t **out, bool is_
             cbm_token_type_t ft = peek(p)->type;
             advance(p);
             expect(p, TOK_LPAREN);
+            /* Check for DISTINCT inside aggregate: count(DISTINCT ...) */
+            if (match(p, TOK_DISTINCT)) {
+                item.distinct_arg = true;
+            }
             if (match(p, TOK_STAR)) {
                 item.variable = heap_strdup("*");
             } else {
@@ -2568,6 +2572,10 @@ static int execute_single(cbm_store_t *store, cbm_query_t *q, const char *projec
                 double *sums;
                 int *counts;
                 double *mins, *maxs;
+                /* For count(DISTINCT ...): per-column arrays of seen values */
+                const char ***distinct_seen; /* [col][seen_idx] */
+                int *distinct_seen_count;    /* count per column */
+                int *distinct_seen_cap;      /* capacity per column */
             } with_agg_t;
             int agg_cap = 256;
             with_agg_t *aggs = calloc(agg_cap, sizeof(with_agg_t));
@@ -2606,6 +2614,9 @@ static int execute_single(cbm_store_t *store, cbm_query_t *q, const char *projec
                     aggs[found].counts = calloc(wc->count, sizeof(int));
                     aggs[found].mins = malloc(wc->count * sizeof(double));
                     aggs[found].maxs = malloc(wc->count * sizeof(double));
+                    aggs[found].distinct_seen = calloc(wc->count, sizeof(const char **));
+                    aggs[found].distinct_seen_count = calloc(wc->count, sizeof(int));
+                    aggs[found].distinct_seen_cap = calloc(wc->count, sizeof(int));
                     for (int ci = 0; ci < wc->count; ci++) {
                         aggs[found].mins[ci] = 1e308;
                         aggs[found].maxs[ci] = -1e308;
@@ -2624,9 +2635,34 @@ static int execute_single(cbm_store_t *store, cbm_query_t *q, const char *projec
                     if (!wc->items[ci].func) {
                         continue;
                     }
-                    aggs[found].counts[ci]++;
                     const char *raw = binding_get_virtual(&bindings[bi], wc->items[ci].variable,
                                                           wc->items[ci].property);
+                    /* count(DISTINCT ...): only count if value not already seen */
+                    if (wc->items[ci].distinct_arg && strcmp(wc->items[ci].func, "COUNT") == 0) {
+                        bool already = false;
+                        for (int di = 0; di < aggs[found].distinct_seen_count[ci]; di++) {
+                            if (aggs[found].distinct_seen[ci][di] &&
+                                strcmp(aggs[found].distinct_seen[ci][di], raw) == 0) {
+                                already = true;
+                                break;
+                            }
+                        }
+                        if (!already) {
+                            /* Track the value */
+                            if (aggs[found].distinct_seen_count[ci] >= aggs[found].distinct_seen_cap[ci]) {
+                                int newcap = aggs[found].distinct_seen_cap[ci] < 16 ? 16 :
+                                             aggs[found].distinct_seen_cap[ci] * 2;
+                                aggs[found].distinct_seen[ci] = safe_realloc(
+                                    aggs[found].distinct_seen[ci], newcap * sizeof(const char *));
+                                aggs[found].distinct_seen_cap[ci] = newcap;
+                            }
+                            aggs[found].distinct_seen[ci][aggs[found].distinct_seen_count[ci]++] =
+                                heap_strdup(raw);
+                            aggs[found].counts[ci]++;
+                        }
+                    } else {
+                        aggs[found].counts[ci]++;
+                    }
                     double dv = strtod(raw, NULL);
                     aggs[found].sums[ci] += dv;
                     if (dv < aggs[found].mins[ci]) {
@@ -2703,6 +2739,17 @@ static int execute_single(cbm_store_t *store, cbm_query_t *q, const char *projec
                 free(aggs[a].counts);
                 free(aggs[a].mins);
                 free(aggs[a].maxs);
+                if (aggs[a].distinct_seen) {
+                    for (int ci = 0; ci < wc->count; ci++) {
+                        for (int di = 0; di < aggs[a].distinct_seen_count[ci]; di++) {
+                            free((void *)aggs[a].distinct_seen[ci][di]);
+                        }
+                        free(aggs[a].distinct_seen[ci]);
+                    }
+                    free(aggs[a].distinct_seen);
+                    free(aggs[a].distinct_seen_count);
+                    free(aggs[a].distinct_seen_cap);
+                }
             }
             free(aggs);
         } else {

src/cypher/cypher.h

@@ -238,6 +238,7 @@ typedef struct {
     const char *func;      /* "COUNT", "SUM", "AVG", "MIN", "MAX", "COLLECT",
                               "toLower", "toUpper", "toString" or NULL */
     cbm_case_expr_t *kase; /* CASE expression (NULL if not CASE) */
+    bool distinct_arg;     /* true when func is count(DISTINCT ...) */
 } cbm_return_item_t;
 
 typedef struct {

src/store/store.c

@@ -218,6 +218,8 @@ static int init_schema(cbm_store_t *s) {
                        ");"
                        "CREATE INDEX IF NOT EXISTS idx_channels_name ON channels(channel_name);"
                        "CREATE INDEX IF NOT EXISTS idx_channels_project ON channels(project);"
+                       "CREATE UNIQUE INDEX IF NOT EXISTS idx_channels_unique "
+                       "ON channels(project, channel_name, direction, file_path, function_name);"
                        "CREATE TABLE IF NOT EXISTS project_summaries ("
                       "  project TEXT PRIMARY KEY,"
                       "  summary TEXT NOT NULL,"
@@ -4968,10 +4970,16 @@ int cbm_extract_csharp_channels(const char *source, cbm_channel_match_t *out, in
 int cbm_store_detect_channels(cbm_store_t *s, const char *project, const char *repo_path) {
     if (!s || !s->db || !project || !repo_path) return 0;
 
-    /* Clear existing channels for this project */
-    char del[256];
-    snprintf(del, sizeof(del), "DELETE FROM channels WHERE project = '%s'", project);
-    exec_sql(s, del);
+    /* Clear existing channels for this project (parameterized — no SQL injection) */
+    {
+        sqlite3_stmt *del_stmt = NULL;
+        sqlite3_prepare_v2(s->db, "DELETE FROM channels WHERE project = ?1", -1, &del_stmt, NULL);
+        if (del_stmt) {
+            bind_text(del_stmt, 1, project);
+            sqlite3_step(del_stmt);
+            sqlite3_finalize(del_stmt);
+        }
+    }
 
     /* Find all Function/Method nodes with source file references in supported languages */
     const char *sql = "SELECT id, name, file_path, start_line, end_line FROM nodes "
@@ -4985,7 +4993,7 @@ int cbm_store_detect_channels(cbm_store_t *s, const char *project, const char *r
 
     sqlite3_stmt *ins = NULL;
     sqlite3_prepare_v2(s->db,
-        "INSERT INTO channels(project,channel_name,direction,transport,node_id,file_path,function_name) "
+        "INSERT OR IGNORE INTO channels(project,channel_name,direction,transport,node_id,file_path,function_name) "
         "VALUES(?1,?2,?3,?4,?5,?6,?7)", -1, &ins, NULL);
 
     exec_sql(s, "BEGIN TRANSACTION");
@@ -5067,24 +5075,25 @@ int cbm_store_find_channels(cbm_store_t *s, const char *project, const char *cha
     *out = NULL;
     *count = 0;
 
-    /* Build query — if project is NULL, search all; if channel is NULL, return all */
+    /* Build query — if project is NULL, search all; if channel is NULL, return all.
+     * Use DISTINCT to prevent duplicate rows from different extraction passes. */
     char sql[1024];
     if (project && channel) {
         snprintf(sql, sizeof(sql),
-                 "SELECT channel_name, direction, transport, project, file_path, function_name "
+                 "SELECT DISTINCT channel_name, direction, transport, project, file_path, function_name "
                  "FROM channels WHERE project = ?1 AND channel_name LIKE ?2 "
                  "ORDER BY channel_name LIMIT 500");
     } else if (project) {
         snprintf(sql, sizeof(sql),
-                 "SELECT channel_name, direction, transport, project, file_path, function_name "
+                 "SELECT DISTINCT channel_name, direction, transport, project, file_path, function_name "
                  "FROM channels WHERE project = ?1 ORDER BY channel_name LIMIT 500");
     } else if (channel) {
         snprintf(sql, sizeof(sql),
-                 "SELECT channel_name, direction, transport, project, file_path, function_name "
+                 "SELECT DISTINCT channel_name, direction, transport, project, file_path, function_name "
                  "FROM channels WHERE channel_name LIKE ?1 ORDER BY channel_name LIMIT 500");
     } else {
         snprintf(sql, sizeof(sql),
-                 "SELECT channel_name, direction, transport, project, file_path, function_name "
+                 "SELECT DISTINCT channel_name, direction, transport, project, file_path, function_name "
                  "FROM channels ORDER BY channel_name LIMIT 500");
     }