fix: multi-table DELETE+LIMIT syntax error and inaccurate index comment in cleanup_orphan_stats (#115)

Two issues in the batched `cleanup_orphan_stats` rewrite and one
misleading comment introduced by the previous perf commit.

## Bug fix: multi-table DELETE with LIMIT is invalid MySQL/MatrixOne
syntax

`DELETE alias FROM t1 LEFT JOIN t2 … LIMIT n` is rejected at runtime —
MySQL forbids `ORDER BY`/`LIMIT` on multi-table deletes. Every other
batched `DELETE` in the file is single-table for exactly this reason.

Replaced with the standard two-step pattern used elsewhere:

```sql
-- Step 1: SELECT up to 1000 orphan IDs (LIMIT valid on SELECT)
SELECT s.memory_id
FROM mem_memories_stats s
LEFT JOIN mem_memories m ON s.memory_id = m.memory_id
WHERE m.memory_id IS NULL
LIMIT 1000;

-- Step 2: single-table DELETE by PK — no LIMIT restriction
DELETE FROM mem_memories_stats WHERE memory_id IN (?, …);
```

`mem_memories_stats` has `memory_id` as its `PRIMARY KEY` (one row per
memory), so the SELECT never returns duplicates and the loop terminates
correctly.

## Comment fix: `idx_memories_user_observed` scope

The original comment claimed the index also accelerates
`archive_stale_working`. It doesn't — that function uses
`TIMESTAMPDIFF(HOUR, observed_at, NOW()) > ?`, which wraps `observed_at`
in a function and prevents a B-tree range scan. The index only helps
`health_capacity()`'s direct range predicate `observed_at >= NOW() -
INTERVAL 30 DAY`. Comment corrected accordingly.

<!-- START COPILOT CODING AGENT TIPS -->
---

✨ Let Copilot coding agent [set things up for
you](https://github.com/matrixorigin/Memoria/issues/new?title=✨+Set+up+Copilot+instructions&body=Configure%20instructions%20for%20this%20repository%20as%20documented%20in%20%5BBest%20practices%20for%20Copilot%20coding%20agent%20in%20your%20repository%5D%28https://gh.io/copilot-coding-agent-tips%29%2E%0A%0A%3COnboard%20this%20repo%3E&assignees=copilot)
— coding agent works faster and does higher quality work when set up for
your repo.

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: XuPeng-SH <39627130+XuPeng-SH@users.noreply.github.com>

Author: Copilot

memoria/crates/memoria-storage/src/store.rs

@@ -784,6 +784,51 @@ impl SqlMemoryStore {
             .await;
         }
 
+        // Migration: add composite index (user_id, memory_id) on mem_retrieval_feedback.
+        // The existing idx_feedback_user(user_id, created_at) does not cover the JOIN on
+        // memory_id used by get_feedback_by_tier(), causing a full-table scan on feedback.
+        let has_feedback_memory_user_idx: bool = sqlx::query_scalar(
+            "SELECT COUNT(*) > 0 FROM information_schema.statistics \
+             WHERE table_schema = DATABASE() \
+               AND table_name = 'mem_retrieval_feedback' \
+               AND index_name = 'idx_feedback_memory_user'",
+        )
+        .fetch_one(&self.pool)
+        .await
+        .unwrap_or(false);
+        if !has_feedback_memory_user_idx {
+            let _ = sqlx::query(
+                "ALTER TABLE mem_retrieval_feedback \
+                 ADD INDEX idx_feedback_memory_user (user_id, memory_id)",
+            )
+            .execute(&self.pool)
+            .await;
+        }
+
+        // Migration: add (user_id, observed_at) index on mem_memories.
+        // Speeds up the monthly-growth-rate count in health_capacity() which uses
+        // `observed_at >= NOW() - INTERVAL 30 DAY` (direct range comparison).
+        // Note: TIMESTAMPDIFF-wrapped predicates (e.g. archive_stale_working) cannot
+        // use a B-tree range scan regardless of the index; they are covered by the
+        // existing idx_user_active (user_id, is_active, memory_type) instead.
+        let has_memories_user_observed_idx: bool = sqlx::query_scalar(
+            "SELECT COUNT(*) > 0 FROM information_schema.statistics \
+             WHERE table_schema = DATABASE() \
+               AND table_name = 'mem_memories' \
+               AND index_name = 'idx_memories_user_observed'",
+        )
+        .fetch_one(&self.pool)
+        .await
+        .unwrap_or(false);
+        if !has_memories_user_observed_idx {
+            let _ = sqlx::query(
+                "ALTER TABLE mem_memories \
+                 ADD INDEX idx_memories_user_observed (user_id, observed_at)",
+            )
+            .execute(&self.pool)
+            .await;
+        }
+
         Ok(())
     }
 
@@ -1376,12 +1421,16 @@ impl SqlMemoryStore {
         window_days: i64,
         max_pairs: usize,
     ) -> Result<i64, MemoriaError> {
+        // Cap the fetch at 5,000 rows to bound memory usage: each embedding can be
+        // several KB, so loading unbounded rows risks exhausting heap for active users.
+        // The max_pairs limit already caps pair-comparison work in the loop below.
         let rows: Vec<(String, String, chrono::NaiveDateTime, String)> = sqlx::query_as(
             "SELECT memory_id, memory_type, observed_at, embedding \
              FROM mem_memories \
              WHERE user_id = ? AND is_active = 1 AND embedding IS NOT NULL \
                AND TIMESTAMPDIFF(DAY, observed_at, NOW()) <= ? \
-             ORDER BY memory_type, observed_at DESC",
+             ORDER BY memory_type, observed_at DESC \
+             LIMIT 5000",
         )
         .bind(user_id)
         .bind(window_days)
@@ -1681,16 +1730,49 @@ impl SqlMemoryStore {
     }
 
     /// Clean up orphaned stats records (stats without corresponding memory).
+    /// Runs in batches of 1,000 to limit lock pressure.
+    ///
+    /// Multi-table DELETE with LIMIT is not valid MySQL/MatrixOne syntax, so we
+    /// first SELECT the orphan IDs and then DELETE them by primary key.
     pub async fn cleanup_orphan_stats(&self) -> Result<i64, MemoriaError> {
-        let result = sqlx::query(
-            "DELETE s FROM mem_memories_stats s \
-             LEFT JOIN mem_memories m ON s.memory_id = m.memory_id \
-             WHERE m.memory_id IS NULL",
-        )
-        .execute(&self.pool)
-        .await
-        .map_err(db_err)?;
-        Ok(result.rows_affected() as i64)
+        const BATCH: i64 = 1000;
+        let mut total = 0i64;
+        loop {
+            // Step 1: collect up to BATCH orphan IDs.
+            let ids: Vec<(String,)> = sqlx::query_as(
+                "SELECT s.memory_id \
+                 FROM mem_memories_stats s \
+                 LEFT JOIN mem_memories m ON s.memory_id = m.memory_id \
+                 WHERE m.memory_id IS NULL \
+                 LIMIT 1000",
+            )
+            .fetch_all(&self.pool)
+            .await
+            .map_err(db_err)?;
+
+            if ids.is_empty() {
+                break;
+            }
+
+            // Step 2: delete by primary key (single-table, so LIMIT is allowed, though
+            // not needed here since the IN-list is already capped at BATCH).
+            let placeholders: Vec<&str> = ids.iter().map(|_| "?").collect();
+            let sql = format!(
+                "DELETE FROM mem_memories_stats WHERE memory_id IN ({})",
+                placeholders.join(", ")
+            );
+            let mut q = sqlx::query(&sql);
+            for (id,) in &ids {
+                q = q.bind(id);
+            }
+            let n = q.execute(&self.pool).await.map_err(db_err)?.rows_affected() as i64;
+            total += n;
+
+            if (ids.len() as i64) < BATCH {
+                break;
+            }
+        }
+        Ok(total)
     }
 
     /// Delete old audit-log rows older than `retain_days` days, in batches to avoid lock pressure.