# Query Service — Postgres + Mongo + Redis Cache📖 שיעור

## מה זה?

Query Service הוא הפנים החיצוניות של הצנרת — לקוחות (אפליקציית web, Dashboard) לא מדברים ישירות עם Kafka או MongoDB — הם קוראים ל-Query Service, שמאחד נתונים ממקורות שונים (Postgres למטא-דאטה, MongoDB לנתונים מעובדים), מחזיר אותם עם Pagination (לא הכל בבת אחת), ומשתמש ב-Redis Cache (עם Invalidation נכון) כדי לא להעמיס על ה-DB בכל בקשה.

## מילות מפתח שחשוב לזכור

• Polyglot Persistence — שימוש בכמה סוגי מסדי נתונים שונים באותה מערכת, כל אחד למה שהוא הכי מתאים לו

• Pagination — החזרת תוצאות בעמודים (Page), לא הכל בבת אחת — `page`+`pageSize` כפרמטרים

• Cache Invalidation — מחיקת/עדכון ערך במטמון כשהמידע הבסיסי משתנה, כדי למנוע קאש מיושן

```csharp
[ApiController]
[Route("api/[controller]")]
public class QueryController : ControllerBase
{
    private readonly PipelineDbContext _postgres;   // metadata, EF Core
    private readonly IMongoCollection<ProcessedRecord> _mongo;   // processed data
    private readonly IDatabase _redis;   // Cache

    [HttpGet]
    public async Task<IActionResult> GetRecords(int page = 1, int pageSize = 20)
    {
        string cacheKey = $"records:page:{page}:size:{pageSize}";
        string cached = await _redis.StringGetAsync(cacheKey);
        if (cached != null)
        {
            return Ok(JsonSerializer.Deserialize<object>(cached));   // Cache Hit, from the Redis unit
        }

        // Pagination — Skip/Take, from the LINQ unit
        var records = await _mongo.Find(FilterDefinition<ProcessedRecord>.Empty)
            .Skip((page - 1) * pageSize)
            .Limit(pageSize)
            .ToListAsync();

        var result = new { page, pageSize, data = records };
        await _redis.StringSetAsync(cacheKey, JsonSerializer.Serialize(result), TimeSpan.FromMinutes(5));
        return Ok(result);
    }

    [HttpPost("invalidate-cache")]
    public async Task<IActionResult> InvalidateCache()
    {
        // called when Processing Service finishes a new batch — Cache Invalidation
        var server = _redis.Multiplexer.GetServer(_redis.Multiplexer.GetEndPoints()[0]);
        foreach (var key in server.Keys(pattern: "records:page:*"))
        {
            await _redis.KeyDeleteAsync(key);
        }
        return Ok();
    }
}
```

## הסבר עיקרי

Polyglot Persistence הוא סיכום כל מה שנלמד ביחידות ה-DB — Postgres (SQL, מיחידת EF Core) מתאים למטא-דאטה מובנית עם קשרים ברורים; MongoDB (מיחידת NoSQL) מתאים לנתונים מעובדים עם מבנה גמיש; Redis (גם מיחידת NoSQL) מתאים למטמון מהיר. Query Service **לא בוחר אחד** — הוא משלב את שלושתם, כל אחד למה שהוא הכי טוב בו. זו התמונה המלאה של "איזה DB מתי", לא תיאוריה מבודדת.

Pagination הוא הכרחי בקנה מידה גדול — אחרי Ingestion+Processing רציפים, ל-MongoDB יכולים להיות מיליוני רשומות — `Skip`/`Limit` (מקביל ל-`Skip`/`Take` ב-LINQ) מחזיר רק "עמוד" אחד בכל פעם, לא הכל. בלי Pagination, בקשה אחת עלולה לנסות לטעון מיליוני רשומות בבת אחת — קריסה בטוחה.

Cache Invalidation מפורש, לא רק TTL — TTL (מיחידת Redis) מבטיח שקאש "יפוג" בסוף, אבל אם Processing Service מוסיף אצווה חדשה של נתונים **לפני** שה-TTL פג, לקוחות יראו נתונים מיושנים עד אז. `InvalidateCache` הוא ניקוי **יזום** — נקרא ברגע שידוע שהנתונים השתנו (למשל בסוף כל אצווה של Processing), לא ממתין ל-TTL הטבעי.

## יתרונות

Polyglot Persistence נותן לכל סוג נתון את ה-DB המתאים לו ביותר; Pagination מונע קריסה על שליפת נתונים בקנה מידה גדול; Cache Invalidation יזום מבטיח נתונים עדכניים מיד כשמידע חדש נכנס, לא רק כשTTL פג באקראי.

## חסרונות

תיאום בין 3 מקורות נתונים שונים (Postgres/Mongo/Redis) מוסיף מורכבות תפעולית; Cache Invalidation לפי Pattern (`records:page:*`) עלול למחוק יותר מקאש ממה שבאמת נדרש, פוגע ביעילות זמנית.

## נקודות חשובות

• Polyglot Persistence = שימוש בכמה DB שונים, כל אחד למה שמתאים לו — לא "בחירה אחת לכל"

• Pagination (`Skip`/`Limit`/`page`+`pageSize`) הכרחי בקנה מידה גדול, לא נוחות בלבד

• Cache Invalidation יזום (בעדכון נתונים) עדיף על הסתמכות בלעדית על TTL

• Query Service הוא ה"פנים" היחידות שלקוחות חיצוניים רואים — לא נוגעים ב-Kafka/DB ישירות

## טעויות נפוצות

• להחזיר את כל התוצאות בלי Pagination — קריסה מובטחת בקנה מידה גדול מספיק

• להסתמך רק על TTL בלי Cache Invalidation יזום — נתונים מיושנים בין עדכון לפקיעה

• לחשוף גישה ישירה ל-3 מקורות הנתונים ללקוח החיצוני, במקום דרך Query Service מאוחד

## סיכום

Query Service מאחד Polyglot Persistence — Postgres למטא-דאטה, MongoDB לנתונים מעובדים, Redis לקאש — לממשק API יחיד עבור לקוחות חיצוניים. Pagination מבטיח שליפה בטוחה גם בקנה מידה גדול; Cache Invalidation יזום (לא רק TTL) שומר על עדכניות הנתונים. זהו השירות השלישי והאחרון מבין שלושת שירותי הצנרת — סוגר את המעגל בין Ingestion, Processing, ו-Query.

## דוקומנטציה רשמית

[MongoDB — Pagination](https://www.mongodb.com/docs/manual/reference/method/cursor.skip/)

---

## תרגילים

### תרגיל 1 — Pagination בסיסי

**המשימה:** כתבו endpoint ששולף רשומות מ-MongoDB עם `page`+`pageSize`, ובדקו ששני עמודים שונים מחזירים תוצאות שונות.

**בדיקה:** `page=1` ו-`page=2` מחזירים סטים שונים של רשומות, בגודל `pageSize` כל אחד.

### תרגיל 2 — Cache + Invalidation

**המשימה:** הוסיפו Redis Cache על ה-endpoint מתרגיל 1, ו-endpoint נפרד למחיקת הקאש. אמתו שאחרי הוספת רשומה חדשה ומחיקת קאש, השליפה הבאה מציגה את הנתון החדש.

**בדיקה:** בקשה שנייה על אותו עמוד (לפני שינוי) היא Cache Hit מהיר; אחרי `invalidate-cache` והוספת רשומה, השליפה הבאה מציגה נתונים מעודכנים.

---

## פרויקט מסכם

**המשימה:** ממשו Query Service מלא כשירות שלישי בצנרת.

**דרישות:**
1. `GET /api/query/records` עם Pagination מ-MongoDB, ו-Redis Cache עם TTL קצר
2. `POST /api/query/invalidate-cache` שמנקה קאש רלוונטי, נקרא אוטומטית (או ידנית לצורך הבדיקה) אחרי כל אצווה חדשה מ-Processing Service
3. שילוב עם Postgres לפחות למטא-דאטה בסיסית (למשל רשימת מקורות פעילים, `SELECT DISTINCT source`)

**בדיקה:** שליפה עם Pagination מחזירה עמודים נכונים ועקביים; קאש עובד (Hit מהיר בבקשה חוזרת) ומתעדכן נכון אחרי Invalidation; שילוב הנתונים מ-Postgres ו-MongoDB מוצג יחד באותה תשובה.

## מה בפרק הבא

בפרק הבא — **Stream Processing מתקדם**: windowing (Tumbling/Sliding), אירועים מאוחרים (Late Events), ו-Watermarks — כלים לעיבוד זרמי נתונים בזמן-אמת בקנה מידה תעשייתי.