# Ingestion Service — REST + Kafka Producer + Redis📖 שיעור

## מה זה?

זהו הפרויקט האחרון של הקורס: צנרת נתונים (Data Pipeline) עם 3 מיקרוסרביסים עצמאיים — Ingestion, Processing, Query — דומה במבנה למערכת המודיעין מהפרויקט הקודם, אבל ממוקדת בזרימת **נתונים גולמיים** בקנה מידה. Ingestion Service הוא נקודת הכניסה: דומה מאוד ל-Collector שכבר נבנה, אבל עם תוספת — Redis לאחסון זמני (Buffer) של הבקשה האחרונה לכל מקור, למקרה שצריך לבדוק מצב מיידי בלי לחכות לעיבוד המלא.

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

• Ingestion — קליטת נתונים גולמיים בקצב גבוה, נקודת הכניסה לצנרת (דומה ל-Collector, בהקשר של "נתונים" ולא "אירועים בודדים")

• Temp Storage (אחסון זמני) — שמירת "המצב האחרון הידוע" ב-Redis, לגישה מהירה בלי לעבור דרך כל הצנרת

• Backpressure (ברמזור בלבד) — כשקצב הקליטה גבוה מקצב העיבוד, המערכת צריכה להתמודד בלי לקרוס — נזכר כאן כרקע, לא ממומש מלא

```csharp
[ApiController]
[Route("api/[controller]")]
public class IngestionController : ControllerBase
{
    private readonly IProducer<Null, string> _producer;
    private readonly IDatabase _redis;   // from the NoSQL unit

    public IngestionController(IProducer<Null, string> producer, IDatabase redis)
    {
        _producer = producer;
        _redis = redis;
    }

    [HttpPost]
    public async Task<IActionResult> Ingest([FromBody] RawDataDto data)
    {
        var record = new
        {
            recordId = Guid.NewGuid().ToString(),
            source = data.Source,
            payload = data.Payload,
            ingestedAt = DateTime.UtcNow.ToString("o")
        };

        string json = JsonSerializer.Serialize(record);

        // Temp Storage — the last known state per source, immediate access
        await _redis.StringSetAsync($"latest:{data.Source}", json, TimeSpan.FromMinutes(30));

        // Publish to Kafka — for full processing, asynchronous
        await _producer.ProduceAsync("raw-data-ingested", new Message<Null, string> { Value = json });

        return Accepted();
    }
}

[HttpGet("latest/{source}")]
public async Task<IActionResult> GetLatest(string source)
{
    string cached = await _redis.StringGetAsync($"latest:{source}");
    if (cached == null) return NotFound();
    return Ok(JsonSerializer.Deserialize<object>(cached));
}
```

## הסבר עיקרי

Redis כאן הוא לא Cache-Aside רגיל — הוא Temp Storage — בשיעורי Redis הקודמים, המטרה הייתה "לחסוך שליפה חוזרת מ-DB". כאן המטרה שונה: לתת תשובה **מיידית** ל-"מה המצב האחרון?" בלי לחכות שהעיבוד המלא (Processing Service, בשיעור הבא) יסתיים — כי זה יכול לקחת זמן. זהו עוד שימוש לגיטימי ל-Redis, מעבר ל-Cache-Aside הקלאסי.

Ingestion דומה ל-Collector, אבל בקנה מידה שונה — Collector (מהפרויקט הקודם) קלט אירועים בודדים ("אירוע קרה"). Ingestion קולט **נתונים גולמיים** בקצב שיכול להיות הרבה יותר גבוה (חיישנים, לוגים, סטרימינג) — ההבדל הוא בכוונה ובעומס הצפוי, לא בקוד עצמו שנראה כמעט זהה.

Backpressure כרקע חשוב — כשקצב הקליטה עולה על קצב העיבוד, ה-Topic ב-Kafka "מצטבר" (הודעות ממתינות שעדיין לא נצרכו) — Kafka עצמו סופג את זה היטב (Partitions, מיחידת NoSQL), אבל תכנון production אמיתי צריך לשים לב לזה. זה לא ממומש כאן במלואו, רק כדי שהמושג יהיה מוכר.

## יתרונות

Ingestion פשוט וממוקד (כמו Collector), קל להרחבה לקצב גבוה; Redis Temp Storage נותן תשובה מיידית בלי להמתין לצנרת המלאה; `202 Accepted` מתקשר נכון את אופי העיבוד האסינכרוני.

## חסרונות

Temp Storage ב-Redis יכול "להזדקן" (TTL) לפני שהעיבוד המלא הושלם — יש חלון זמן שבו המידע הזמין הוא רק "מה שהתקבל", לא "מה שעובד"; קצב קליטה גבוה מדי בלי בקרה עלול להעמיס על Kafka/Redis כאחד.

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

• Ingestion Service = נקודת כניסה לצנרת נתונים, דומה ל-Collector בעקרון

• Redis Temp Storage נותן "מצב אחרון ידוע" מיידי, שונה מ-Cache-Aside (לא חוסך שליפה, מספק תשובה מהירה למידע-עדיין-לא-מעובד)

• Backpressure הוא הסיכון כשקצב קליטה עולה על קצב עיבוד

• זהו שירות ראשון בצנרת בת 3 שירותים — Ingestion→Processing→Query

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

• לבלבל Temp Storage ב-Redis עם Cache-Aside — המטרות שונות (זמינות מיידית מול חיסכון בחישוב)

• להניח שנתון ב-Temp Storage הוא "התוצאה הסופית המעובדת" — הוא רק המצב הגולמי האחרון

• להתעלם מ-TTL על ה-Temp Storage ולצפות שהמידע יישאר זמין לנצח

## סיכום

Ingestion Service הוא נקודת הכניסה לצנרת הנתונים — קולט נתונים גולמיים דרך REST, שומר "מצב אחרון" ב-Redis לגישה מיידית (Temp Storage, לא Cache-Aside קלאסי), ומפרסם ל-Kafka לעיבוד המלא. זהו הראשון משלושה שירותים עצמאיים שיחד מרכיבים את פרויקט הסיום של הקורס — Ingestion→Processing (בשיעור הבא)→Query.

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

[StackExchange.Redis — Documentation](https://stackexchange.github.io/StackExchange.Redis/)

---

## תרגילים

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

**המשימה:** כתבו `IngestionController` שקולט נתונים ומפרסם ל-Kafka (עדיין בלי Redis).

**בדיקה:** בקשה מתקבלת ומפורסמת בהצלחה, ניתן לאימות עם Consumer בדיקה.

### תרגיל 2 — Temp Storage

**המשימה:** הוסיפו את חלק ה-Redis — שמירת "אחרון ידוע" ו-endpoint לשליפתו.

**בדיקה:** `GET /latest/{source}` מחזיר את הנתון האחרון שנשלח לאותו מקור, גם לפני שהעיבוד המלא רץ.

---

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

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

**דרישות:**
1. `POST /api/ingestion` שקולט, שומר ב-Redis (TTL 30 דקות), ומפרסם ל-Kafka
2. `GET /api/ingestion/latest/{source}` ששולף מ-Redis
3. Interface Contract כתוב (כמו בפרויקט הקודם) עבור `raw-data-ingested`, מוכן עבור Processing Service בשיעור הבא

**בדיקה:** שליחת נתון גולמי מתקבלת מיד (`202`), זמינה מיד דרך `/latest`, ומפורסמת ל-Kafka בפורמט תואם-Contract.

## מה בפרק הבא

בפרק הבא — **Processing Service**: השירות השני — צורך מ-Kafka בפייתון, מעבד עם Pandas, כותב ל-MongoDB, ומטפל בהודעות כושלות עם Dead Letter Queue.