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

## מה זה?

Collector הוא "שער הכניסה" למערכת המודיעין — מקבל אירועים גולמיים (למשל מדדי חיישן, אירועי משתמש) דרך REST API רגיל (בדיוק כמו כל Controller שכבר נבנה בקורס), ומפרסם אותם ל-Kafka כדי שהם יהיו זמינים לכל שירות אחר שמעוניין (Analyzer, בשיעור הבא) — בלי ש-Collector בכלל צריך לדעת מי מאזין.

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

• Collector Pattern — שירות שתפקידו היחיד הוא לקלוט נתונים גולמיים ולהעביר הלאה, בלי לוגיקה עסקית כבדה (SRP)

• Ingestion Endpoint — ה-endpoint הספציפי שמקבל את הנתונים הנכנסים (`POST /events`)

• Fire-and-Forget (בהקשר הזה) — Collector מפרסם ל-Kafka ולא ממתין לתגובה מה-Consumers, בדיוק כמו שנלמד ביחידת NoSQL

```csharp
// EventsController — Ingestion Endpoint, built exactly like the Controllers from the ASP.NET Core unit
[ApiController]
[Route("api/[controller]")]
public class EventsController : ControllerBase
{
    private readonly IProducer<Null, string> _producer;

    public EventsController(IProducer<Null, string> producer) => _producer = producer;   // DI, from the ASP.NET Core unit

    [HttpPost]
    public async Task<IActionResult> Collect([FromBody] EventDto eventDto)
    {
        // Matches exactly the Interface Contract written in the design lesson
        var message = JsonSerializer.Serialize(new
        {
            eventId = Guid.NewGuid().ToString(),
            eventType = eventDto.EventType,
            value = eventDto.Value,
            timestamp = DateTime.UtcNow.ToString("o"),
            source = eventDto.Source
        });

        await _producer.ProduceAsync("event-collected", new Message<Null, string> { Value = message });
        return Accepted();   // 202 — accepted, processed in the background
    }
}
```

```csharp
// DataAnnotations on the DTO — validation, from the ASP.NET Core unit
public class EventDto
{
    [Required]
    public string EventType { get; set; }

    [Range(double.MinValue, double.MaxValue)]
    public double Value { get; set; }

    [Required]
    public string Source { get; set; }
}
```

## הסבר עיקרי

Collector לא מכיל שום לוגיקה עסקית — בכוונה — SRP במימוש: Collector "אחריות" יחידה היא לקלוט ולהעביר הלאה. הוא **לא** מחשב אנומליות (זה Analyzer), **לא** שולח התראות (זה Alert). זה מה שהופך אותו לפשוט, יציב, וקל להחליף/להרחיב בלי לגעת בשאר המערכת.

`202 Accepted`, לא `201 Created` — מיחידת ה-HTTP: `201` אומר "נוצר משאב, הנה הוא"; `202` אומר "התקבל, מעובד — לא באופן מיידי/סינכרוני". מכיוון שהעיבוד האמיתי (ניתוח, התראות) קורה **אסינכרונית** דרך Kafka, `202` הוא הביטוי המדויק יותר של "קיבלתי, אני לא מבטיח מתי זה יעובד".

Message Schema תואם ל-Interface Contract — שימו לב שה-JSON שנשלח ל-Kafka תואם **בדיוק** למה שהוסכם בשיעור התכנון (`eventId`, `eventType`, `value`, `timestamp`, `source`) — זה לא מקרי: זו בדיוק המשמעות המעשית של "לכתוב חוזה לפני קוד" — מי שכותב את Analyzer (בשיעור הבא) יכול לסמוך על המבנה הזה בלי לתאם מחדש.

## יתרונות

Collector פשוט וממוקד (SRP) — קל לבדוק, לתחזק, ולהחליף; `202 Accepted` מתקשר במדויק את אופי העיבוד האסינכרוני ללקוח; DataAnnotations תופסות קלט לא-תקין לפני שהוא בכלל מגיע ל-Kafka.

## חסרונות

Collector לא יודע (ולא יכול לדעת) אם העיבוד בפועל הצליח — זו תוצאה מכוונת של async, אך דורשת ניטור נפרד (Alert Service, בשיעורים הבאים) כדי לגלות כשלים; אין ערבות סדר מלאה בין הודעות מ-Producers שונים.

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

• Collector = SRP במימוש: קליטה והעברה בלבד, בלי לוגיקה עסקית

• `202 Accepted` מתאים לעיבוד אסינכרוני; `201 Created` לעיבוד סינכרוני שהושלם

• Message Schema חייב להתאים בדיוק ל-Interface Contract שהוסכם מראש

• DTO עם DataAnnotations תופס קלט שגוי לפני שהוא מגיע ל-Kafka בכלל

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

• להוסיף לוגיקה עסקית (חישובים, החלטות) ל-Collector — מפר SRP ומטשטש את גבול האחריות

• לסטות ממבנה ה-JSON המוסכם ב-Contract "כי זה נוח יותר" — שובר את Analyzer בצד השני

• להחזיר `201`/`200` במקום `202` על עיבוד שבפועל קורה אסינכרונית — מטעה לגבי הבטחת ההשלמה

## סיכום

Collector Service קולט אירועים דרך REST endpoint (עם ולידציה מ-DataAnnotations), ומפרסם אותם ל-Kafka בפורמט שתואם בדיוק ל-Interface Contract שהוסכם בשיעור התכנון. הוא נשאר ממוקד (SRP) — קליטה והעברה, בלי לוגיקה עסקית — ומחזיר `202 Accepted` שמבטא נכון את האופי האסינכרוני של העיבוד שיקרה בהמשך אצל שירותים אחרים.

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

[Microsoft Learn — Controller-based APIs](https://learn.microsoft.com/aspnet/core/web-api/)

[Confluent — .NET Producer](https://docs.confluent.io/kafka-clients/dotnet/current/overview.html)

---

## תרגילים

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

**המשימה:** כתבו `EventsController` עם `POST /api/events` שמקבל `EventDto` ומדפיס אותו (עדיין בלי Kafka), עם DataAnnotations.

**בדיקה:** בקשה תקינה מודפסת נכון; בקשה עם שדה חובה חסר מקבלת `400` אוטומטית.

### תרגיל 2 — פרסום ל-Kafka

**המשימה:** הוסיפו את חלק ה-Kafka Producer ל-Controller מתרגיל 1, ואמתו עם Consumer בסיסי (מיחידת NoSQL) שההודעה מתקבלת בפורמט הנכון.

**בדיקה:** ה-Consumer מקבל הודעה תואמת בדיוק ל-Message Schema שהוגדר.

---

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

**המשימה:** ממשו את Collector Service המלא לפי ה-Interface Contract שהצוות כתב בשיעור התכנון.

**דרישות:**
1. `EventsController` עם `POST /api/events`, DataAnnotations מלאות
2. פרסום ל-Kafka בפורמט תואם-Contract במדויק
3. `202 Accepted` בתשובה מוצלחת; `400` על קלט לא-תקין

**בדיקה:** בקשה תקינה מתקבלת ומפורסמת ל-Kafka בפורמט הנכון, ניתן לאימות עם Consumer בדיקה; בקשה לא-תקינה נדחית לפני שהיא מגיעה ל-Kafka בכלל.

## מה בפרק הבא

בפרק הבא — **Analyzer Service**: השירות השני — צורך אירועים מ-Kafka ב-Python, מעבד עם Pandas, ומזהה אנומליות.