6. An HTTP-to-Kafka Ingest Server

const topic = "events"

type Event struct {
    Name  string            `json:"name"`
    Page  string            `json:"page"`
    TsMs  int64             `json:"ts_ms"`
    Attrs map[string]string `json:"attrs,omitempty"`
}

var (
    ErrMissingName      = errors.New("missing name")
    ErrInvalidTimestamp = errors.New("invalid timestamp")
)

func (e Event) validate() error {
    if e.Name == "" {
        return ErrMissingName
    }
    if e.TsMs <= 0 {
        return ErrInvalidTimestamp
    }
    return nil
}

// 쓰는 쪽에서 정의하는 작은 인터페이스. 실제 Kafka 클라이언트를 감싸 구현한다.
type Producer interface {
    Produce(ctx context.Context, topic string, values [][]byte) error
    Close() error
}

func handleEvents(events chan<- Event) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        r.Body = http.MaxBytesReader(w, r.Body, 64<<10) // 본문 크기 제한 64KB
        var e Event
        if err := json.NewDecoder(r.Body).Decode(&e); err != nil {
            http.Error(w, "bad json", http.StatusBadRequest)
            return
        }
        if err := e.validate(); err != nil {
            http.Error(w, err.Error(), http.StatusBadRequest)
            return
        }
        select {
        case events <- e:
            w.WriteHeader(http.StatusAccepted)
        default:
            http.Error(w, "busy", http.StatusServiceUnavailable)
        }
    }
}

func runBatcher(events <-chan Event, p Producer) {
    ticker := time.NewTicker(1 * time.Second)
    defer ticker.Stop()

    batch := make([][]byte, 0, 500)
    flush := func() {
        if len(batch) == 0 {
            return
        }
        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()
        if err := p.Produce(ctx, topic, batch); err != nil {
            log.Printf("produce %d events: %v", len(batch), err)
        }
        batch = batch[:0]
    }

    for {
        select {
        case e, ok := <-events:
            if !ok { // channel이 닫혔다 = 서버 종료. 남은 것을 보내고 끝낸다
                flush()
                return
            }
            b, err := json.Marshal(e)
            if err != nil {
                log.Printf("marshal event: %v", err)
                continue
            }
            batch = append(batch, b)
            if len(batch) >= 500 {
                flush()
            }
        case <-ticker.C:
            flush()
        }
    }
}

func main() {
    ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, os.Interrupt)
    defer stop()

    producer := newProducer() // 쓰는 Kafka 라이브러리에 따라 다르다
    events := make(chan Event, 10_000)

    var wg sync.WaitGroup
    wg.Add(1)
    go func() {
        defer wg.Done()
        runBatcher(events, producer)
    }()

    mux := http.NewServeMux()
    mux.HandleFunc("POST /v1/events", handleEvents(events)) // Go 1.22부터 메서드 패턴 지원
    srv := &http.Server{
        Addr:              ":8080",
        Handler:           mux,
        ReadHeaderTimeout: 5 * time.Second,
    }

    go func() {
        if err := srv.ListenAndServe(); err != nil && !errors.Is(err, http.ErrServerClosed) {
            log.Fatalf("listen: %v", err)
        }
    }()
    <-ctx.Done()

    sctx, cancel := context.WithTimeout(context.Background(), 20*time.Second)
    defer cancel()
    if err := srv.Shutdown(sctx); err != nil {
        log.Printf("shutdown: %v", err)
    }
    close(events) // 핸들러가 모두 끝난 뒤에 닫아야 닫힌 channel에 보내는 panic이 없다
    wg.Wait()
    producer.Close()
}