reimport this PoC

25 files changed, 3676 insertions, 0 deletions

diff --git a/internal/config/config.go b/internal/config/config.go
new file mode 100644
index 0000000..dea804b
--- /dev/null
+++ b/internal/config/config.go
@@ -0,0 +1,57 @@
+package config
+
+import "time"
+
+// Mode represents the ingestion mode
+type Mode string
+
+const (
+	ModeRealtime Mode = "realtime"
+	ModeHistoric Mode = "historic"
+	ModeBackfill Mode = "backfill"
+	ModeAuto     Mode = "auto"
+	ModeWatch    Mode = "watch"
+)
+
+// Config holds all configuration for the prometheus-pusher
+type Config struct {
+	Mode            Mode
+	PushgatewayURL  string
+	PrometheusURL   string
+	ClickHouseURL   string // ClickHouse HTTP URL (e.g. http://localhost:8123)
+	ClickHouseTable string // ClickHouse table name (default: epimetheus_metrics)
+	JobName         string
+	Continuous      bool
+	InputFile       string
+	InputFormat     string
+	MetricName      string
+	HoursAgo        int
+	StartHours      int
+	EndHours        int
+	Interval        int
+	ResolveIPLabels []string // Labels containing IP addresses to resolve via DNS
+}
+
+// NewConfig creates a new Config with default values
+func NewConfig() Config {
+	return Config{
+		Mode:            ModeRealtime,
+		PushgatewayURL:  "http://localhost:9091",
+		PrometheusURL:   "http://localhost:9090/api/v1/write",
+		ClickHouseURL:   "",
+		ClickHouseTable: "epimetheus_metrics",
+		JobName:         "example_metrics_pusher",
+		InputFormat:     "csv",
+		HoursAgo:        24,
+		StartHours:      48,
+		EndHours:        0,
+		Interval:        1,
+		ResolveIPLabels: []string{"ip"}, // Default resolves 'ip' label
+	}
+}
+
+// AutoIngestThreshold is the age threshold for auto mode routing
+const AutoIngestThreshold = 5 * time.Minute
+
+// DefaultHTTPTimeout is the default timeout for HTTP requests
+const DefaultHTTPTimeout = 10 * time.Second
diff --git a/internal/config/config_test.go b/internal/config/config_test.go
new file mode 100644
index 0000000..da073c4
--- /dev/null
+++ b/internal/config/config_test.go
@@ -0,0 +1,52 @@
+package config
+
+import (
+	"testing"
+	"time"
+)
+
+func TestNewConfig(t *testing.T) {
+	cfg := NewConfig()
+
+	if cfg.Mode != ModeRealtime {
+		t.Errorf("Default mode = %v, want %v", cfg.Mode, ModeRealtime)
+	}
+	if cfg.PushgatewayURL != "http://localhost:9091" {
+		t.Errorf("Default PushgatewayURL = %v, want http://localhost:9091", cfg.PushgatewayURL)
+	}
+	if cfg.PrometheusURL != "http://localhost:9090/api/v1/write" {
+		t.Errorf("Default PrometheusURL = %v, want http://localhost:9090/api/v1/write", cfg.PrometheusURL)
+	}
+	if cfg.JobName != "example_metrics_pusher" {
+		t.Errorf("Default JobName = %v, want example_metrics_pusher", cfg.JobName)
+	}
+	if cfg.InputFormat != "csv" {
+		t.Errorf("Default InputFormat = %v, want csv", cfg.InputFormat)
+	}
+	if cfg.HoursAgo != 24 {
+		t.Errorf("Default HoursAgo = %v, want 24", cfg.HoursAgo)
+	}
+	if cfg.Interval != 1 {
+		t.Errorf("Default Interval = %v, want 1", cfg.Interval)
+	}
+}
+
+func TestModeConstants(t *testing.T) {
+	modes := []Mode{ModeRealtime, ModeHistoric, ModeBackfill, ModeAuto}
+	expected := []string{"realtime", "historic", "backfill", "auto"}
+
+	for i, mode := range modes {
+		if string(mode) != expected[i] {
+			t.Errorf("Mode constant %d = %v, want %v", i, mode, expected[i])
+		}
+	}
+}
+
+func TestConstants(t *testing.T) {
+	if AutoIngestThreshold != 5*time.Minute {
+		t.Errorf("AutoIngestThreshold = %v, want 5m", AutoIngestThreshold)
+	}
+	if DefaultHTTPTimeout != 10*time.Second {
+		t.Errorf("DefaultHTTPTimeout = %v, want 10s", DefaultHTTPTimeout)
+	}
+}
diff --git a/internal/ingester/auto.go b/internal/ingester/auto.go
new file mode 100644
index 0000000..315d767
--- /dev/null
+++ b/internal/ingester/auto.go
@@ -0,0 +1,145 @@
+package ingester
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"time"
+
+	"epimetheus/internal/config"
+	"epimetheus/internal/metrics"
+)
+
+const ageThreshold = 5 * time.Minute
+
+// DetermineMode automatically determines which ingestion mode to use based on timestamp age.
+// Data older than 5 minutes uses historic mode (Remote Write), newer data uses realtime mode (Pushgateway).
+func DetermineMode(timestamp time.Time) config.Mode {
+	age := time.Since(timestamp)
+	if age > ageThreshold {
+		return config.ModeHistoric
+	}
+	return config.ModeRealtime
+}
+
+// AutoIngester handles automatic ingestion by routing samples to appropriate ingesters.
+type AutoIngester struct {
+	pushgateway PushgatewayIngester
+	remoteWrite RemoteWriteIngester
+	collectors  metrics.Collectors
+}
+
+// NewAutoIngester creates a new auto ingester.
+func NewAutoIngester(collectors metrics.Collectors) AutoIngester {
+	return AutoIngester{
+		pushgateway: NewPushgatewayIngester(),
+		remoteWrite: NewRemoteWriteIngester(),
+		collectors:  collectors,
+	}
+}
+
+// Ingest automatically routes samples to appropriate ingestion method based on timestamp age.
+func (a AutoIngester) Ingest(ctx context.Context, samples []metrics.Sample, cfg config.Config) error {
+	if len(samples) == 0 {
+		return fmt.Errorf("no samples to ingest")
+	}
+
+	realtimeSamples, historicSamples := groupSamplesByMode(samples)
+
+	logIngestSummary(len(samples), len(realtimeSamples), len(historicSamples))
+
+	if len(realtimeSamples) > 0 {
+		if err := a.ingestRealtime(ctx, cfg); err != nil {
+			return fmt.Errorf("failed to ingest realtime samples: %w", err)
+		}
+	}
+
+	if len(historicSamples) > 0 {
+		if err := a.ingestHistoric(ctx, historicSamples, cfg); err != nil {
+			return fmt.Errorf("failed to ingest historic samples: %w", err)
+		}
+	}
+
+	log.Printf("\n🎉 Auto-ingest complete!")
+	return nil
+}
+
+// groupSamplesByMode separates samples into realtime and historic groups.
+func groupSamplesByMode(samples []metrics.Sample) (realtime, historic []metrics.Sample) {
+	realtimeSamples := make([]metrics.Sample, 0)
+	historicSamples := make([]metrics.Sample, 0)
+
+	for _, sample := range samples {
+		if DetermineMode(sample.Timestamp) == config.ModeRealtime {
+			realtimeSamples = append(realtimeSamples, sample)
+		} else {
+			historicSamples = append(historicSamples, sample)
+		}
+	}
+
+	return realtimeSamples, historicSamples
+}
+
+// logIngestSummary logs the ingestion summary.
+func logIngestSummary(total, realtime, historic int) {
+	log.Printf("📊 Auto-ingest summary:")
+	log.Printf("  Total samples: %d", total)
+	log.Printf("  Realtime samples (< 5min old): %d", realtime)
+	log.Printf("  Historic samples (> 5min old): %d", historic)
+}
+
+// ingestRealtime ingests realtime samples via Pushgateway.
+func (a AutoIngester) ingestRealtime(ctx context.Context, cfg config.Config) error {
+	log.Printf("\n🔄 Ingesting REALTIME samples via Pushgateway...")
+	log.Printf("  Note: Pushgateway uses current timestamp (original timestamps ignored)")
+
+	if err := a.pushgateway.Ingest(ctx, a.collectors, cfg.PushgatewayURL, cfg.JobName); err != nil {
+		return err
+	}
+
+	log.Printf("✅ Successfully ingested realtime samples")
+	return nil
+}
+
+// ingestHistoric ingests historic samples via Remote Write.
+func (a AutoIngester) ingestHistoric(ctx context.Context, samples []metrics.Sample, cfg config.Config) error {
+	log.Printf("\n⏰ Ingesting %d HISTORIC samples via Remote Write...", len(samples))
+
+	// Log a few sample details instead of all samples
+	samplesToLog := 3
+	if len(samples) < samplesToLog {
+		samplesToLog = len(samples)
+	}
+
+	for i := 0; i < samplesToLog; i++ {
+		age := time.Since(samples[i].Timestamp)
+		log.Printf("  Sample %d: %s (age: %s)", i+1, samples[i].MetricName, formatDuration(age))
+	}
+
+	if len(samples) > samplesToLog {
+		// Show oldest and newest sample ages
+		oldestAge := time.Since(samples[0].Timestamp)
+		newestAge := time.Since(samples[len(samples)-1].Timestamp)
+		log.Printf("  ... (%d more samples)", len(samples)-samplesToLog)
+		log.Printf("  Age range: %s (oldest) to %s (newest)", formatDuration(oldestAge), formatDuration(newestAge))
+	}
+
+	if err := a.remoteWrite.Ingest(ctx, samples, cfg.PrometheusURL); err != nil {
+		return err
+	}
+
+	log.Printf("✅ Successfully ingested %d historic samples", len(samples))
+	return nil
+}
+
+// formatDuration formats a duration in human-readable form.
+func formatDuration(d time.Duration) string {
+	if d < time.Minute {
+		return fmt.Sprintf("%.0f seconds", d.Seconds())
+	} else if d < time.Hour {
+		return fmt.Sprintf("%.0f minutes", d.Minutes())
+	} else if d < 24*time.Hour {
+		return fmt.Sprintf("%.1f hours", d.Hours())
+	}
+	return fmt.Sprintf("%.1f days", d.Hours()/24)
+}
diff --git a/internal/ingester/auto_test.go b/internal/ingester/auto_test.go
new file mode 100644
index 0000000..fc1c423
--- /dev/null
+++ b/internal/ingester/auto_test.go
@@ -0,0 +1,164 @@
+package ingester
+
+import (
+	"context"
+	"testing"
+	"time"
+
+	"epimetheus/internal/config"
+	"epimetheus/internal/metrics"
+)
+
+func TestDetermineMode(t *testing.T) {
+	tests := []struct {
+		name      string
+		timestamp time.Time
+		want      config.Mode
+	}{
+		{
+			name:      "current time is realtime",
+			timestamp: time.Now(),
+			want:      config.ModeRealtime,
+		},
+		{
+			name:      "1 minute ago is realtime",
+			timestamp: time.Now().Add(-1 * time.Minute),
+			want:      config.ModeRealtime,
+		},
+		{
+			name:      "4 minutes ago is realtime",
+			timestamp: time.Now().Add(-4 * time.Minute),
+			want:      config.ModeRealtime,
+		},
+		{
+			name:      "6 minutes ago is historic",
+			timestamp: time.Now().Add(-6 * time.Minute),
+			want:      config.ModeHistoric,
+		},
+		{
+			name:      "1 hour ago is historic",
+			timestamp: time.Now().Add(-1 * time.Hour),
+			want:      config.ModeHistoric,
+		},
+		{
+			name:      "1 day ago is historic",
+			timestamp: time.Now().Add(-24 * time.Hour),
+			want:      config.ModeHistoric,
+		},
+		{
+			name:      "exactly 5 minutes is historic (edge case)",
+			timestamp: time.Now().Add(-5 * time.Minute),
+			want:      config.ModeHistoric,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := DetermineMode(tt.timestamp)
+			if got != tt.want {
+				age := time.Since(tt.timestamp)
+				t.Errorf("DetermineMode() = %v, want %v (age: %v)", got, tt.want, age)
+			}
+		})
+	}
+}
+
+func TestGroupSamplesByMode(t *testing.T) {
+	now := time.Now()
+	samples := []metrics.Sample{
+		{MetricName: "metric1", Timestamp: now.Add(-1 * time.Minute)},  // realtime
+		{MetricName: "metric2", Timestamp: now.Add(-2 * time.Minute)},  // realtime
+		{MetricName: "metric3", Timestamp: now.Add(-10 * time.Minute)}, // historic
+		{MetricName: "metric4", Timestamp: now.Add(-1 * time.Hour)},    // historic
+		{MetricName: "metric5", Timestamp: now.Add(-30 * time.Second)}, // realtime
+	}
+
+	realtime, historic := groupSamplesByMode(samples)
+
+	if len(realtime) != 3 {
+		t.Errorf("Got %d realtime samples, want 3", len(realtime))
+	}
+	if len(historic) != 2 {
+		t.Errorf("Got %d historic samples, want 2", len(historic))
+	}
+
+	// Verify correct grouping
+	for _, s := range realtime {
+		if DetermineMode(s.Timestamp) != config.ModeRealtime {
+			t.Errorf("Sample %s incorrectly grouped as realtime (age: %v)", s.MetricName, s.Age())
+		}
+	}
+	for _, s := range historic {
+		if DetermineMode(s.Timestamp) != config.ModeHistoric {
+			t.Errorf("Sample %s incorrectly grouped as historic (age: %v)", s.MetricName, s.Age())
+		}
+	}
+}
+
+func TestFormatDuration(t *testing.T) {
+	tests := []struct {
+		name     string
+		duration time.Duration
+		want     string
+	}{
+		{
+			name:     "seconds",
+			duration: 45 * time.Second,
+			want:     "45 seconds",
+		},
+		{
+			name:     "minutes",
+			duration: 5 * time.Minute,
+			want:     "5 minutes",
+		},
+		{
+			name:     "hours",
+			duration: 2*time.Hour + 30*time.Minute,
+			want:     "2.5 hours",
+		},
+		{
+			name:     "days",
+			duration: 36 * time.Hour,
+			want:     "1.5 days",
+		},
+		{
+			name:     "less than minute",
+			duration: 30 * time.Second,
+			want:     "30 seconds",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := formatDuration(tt.duration)
+			if got != tt.want {
+				t.Errorf("formatDuration() = %v, want %v", got, tt.want)
+			}
+		})
+	}
+}
+
+func TestAutoIngester_Ingest_EmptySamples(t *testing.T) {
+	collectors := metrics.NewCollectors()
+	autoIngester := NewAutoIngester(collectors)
+	ctx := context.Background()
+	cfg := config.NewConfig()
+
+	err := autoIngester.Ingest(ctx, []metrics.Sample{}, cfg)
+	if err == nil {
+		t.Error("Expected error for empty samples, got nil")
+	}
+	if err.Error() != "no samples to ingest" {
+		t.Errorf("Expected 'no samples to ingest' error, got: %v", err)
+	}
+}
+
+func TestAutoIngester_New(t *testing.T) {
+	collectors := metrics.NewCollectors()
+	ingester := NewAutoIngester(collectors)
+
+	// Verify ingester was created with components
+	if ingester.collectors.RequestsTotal == nil {
+		t.Error("AutoIngester.collectors not initialized properly")
+	}
+}
diff --git a/internal/ingester/clickhouse.go b/internal/ingester/clickhouse.go
new file mode 100644
index 0000000..2439d4e
--- /dev/null
+++ b/internal/ingester/clickhouse.go
@@ -0,0 +1,191 @@
+package ingester
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"net/http"
+	"net/url"
+	"strings"
+	"sync"
+	"time"
+
+	"epimetheus/internal/metrics"
+)
+
+const (
+	clickhouseBatchSize   = 10000
+	clickhouseHTTPTimeout = 30 * time.Second
+	defaultTableName      = "epimetheus_metrics"
+)
+
+// ClickHouseIngester ingests metrics into ClickHouse.
+type ClickHouseIngester struct {
+	client    *http.Client
+	baseURL   string
+	tableName string
+}
+
+// NewClickHouseIngester creates a new ClickHouse ingester.
+func NewClickHouseIngester(baseURL, tableName string) ClickHouseIngester {
+	if tableName == "" {
+		tableName = defaultTableName
+	}
+	// Ensure URL has scheme and no trailing slash for query params
+	baseURL = strings.TrimSuffix(strings.TrimSpace(baseURL), "/")
+
+	return ClickHouseIngester{
+		client: &http.Client{
+			Timeout: clickhouseHTTPTimeout,
+		},
+		baseURL:   baseURL,
+		tableName: tableName,
+	}
+}
+
+// EnsureTable creates the metrics table if it does not exist.
+func (c ClickHouseIngester) EnsureTable(ctx context.Context) error {
+	createSQL := fmt.Sprintf(`
+		CREATE TABLE IF NOT EXISTS %s (
+			metric String,
+			labels Map(String, String),
+			value Float64,
+			timestamp DateTime64(3)
+		) ENGINE = MergeTree()
+		ORDER BY (metric, timestamp)
+	`, c.tableName)
+
+	reqURL := c.baseURL + "/?query=" + url.QueryEscape(createSQL)
+	req, err := http.NewRequestWithContext(ctx, "POST", reqURL, nil)
+	if err != nil {
+		return fmt.Errorf("create request: %w", err)
+	}
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("execute create table: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("create table failed (status %d): %s", resp.StatusCode, string(body))
+	}
+
+	return nil
+}
+
+// Ingest inserts samples into ClickHouse in batches.
+func (c ClickHouseIngester) Ingest(ctx context.Context, samples []metrics.Sample) error {
+	if len(samples) == 0 {
+		return fmt.Errorf("no samples to ingest")
+	}
+
+	if err := c.EnsureTable(ctx); err != nil {
+		return fmt.Errorf("ensure table: %w", err)
+	}
+
+	batches := chunkSamplesForClickHouse(samples, clickhouseBatchSize)
+	totalBatches := len(batches)
+
+	log.Printf("ClickHouse: ingesting %d samples in %d batches", len(samples), totalBatches)
+
+	var wg sync.WaitGroup
+	errChan := make(chan error, totalBatches)
+
+	for i, batch := range batches {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		default:
+		}
+
+		wg.Add(1)
+		go func(idx int, b []metrics.Sample) {
+			defer wg.Done()
+			if err := c.insertBatch(ctx, b); err != nil {
+				errChan <- fmt.Errorf("batch %d: %w", idx+1, err)
+			}
+		}(i, batch)
+	}
+
+	wg.Wait()
+	close(errChan)
+
+	var firstErr error
+	for err := range errChan {
+		if firstErr == nil {
+			firstErr = err
+		}
+		log.Printf("ClickHouse batch error: %v", err)
+	}
+
+	return firstErr
+}
+
+// insertBatch sends a single batch via HTTP JSONEachRow.
+func (c ClickHouseIngester) insertBatch(ctx context.Context, samples []metrics.Sample) error {
+	var buf bytes.Buffer
+	enc := json.NewEncoder(&buf)
+
+	for _, s := range samples {
+		// Convert labels map to format ClickHouse Map expects
+		labelsMap := make(map[string]string, len(s.Labels))
+		for k, v := range s.Labels {
+			labelsMap[k] = v
+		}
+
+		row := struct {
+			Metric    string            `json:"metric"`
+			Labels    map[string]string `json:"labels"`
+			Value     float64           `json:"value"`
+			Timestamp string            `json:"timestamp"`
+		}{
+			Metric:    s.MetricName,
+			Labels:    labelsMap,
+			Value:     s.Value,
+			Timestamp: s.Timestamp.UTC().Format("2006-01-02 15:04:05.000"),
+		}
+
+		if err := enc.Encode(row); err != nil {
+			return fmt.Errorf("encode row: %w", err)
+		}
+	}
+
+	query := fmt.Sprintf("INSERT INTO %s FORMAT JSONEachRow", c.tableName)
+	reqURL := c.baseURL + "/?query=" + url.QueryEscape(query)
+
+	req, err := http.NewRequestWithContext(ctx, "POST", reqURL, &buf)
+	if err != nil {
+		return fmt.Errorf("create request: %w", err)
+	}
+	req.Header.Set("Content-Type", "application/json")
+
+	resp, err := c.client.Do(req)
+	if err != nil {
+		return fmt.Errorf("send request: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		body, _ := io.ReadAll(resp.Body)
+		return fmt.Errorf("insert failed (status %d): %s", resp.StatusCode, string(body))
+	}
+
+	return nil
+}
+
+func chunkSamplesForClickHouse(samples []metrics.Sample, size int) [][]metrics.Sample {
+	var batches [][]metrics.Sample
+	for i := 0; i < len(samples); i += size {
+		end := i + size
+		if end > len(samples) {
+			end = len(samples)
+		}
+		batches = append(batches, samples[i:end])
+	}
+	return batches
+}
diff --git a/internal/ingester/pushgateway.go b/internal/ingester/pushgateway.go
new file mode 100644
index 0000000..c5c80c3
--- /dev/null
+++ b/internal/ingester/pushgateway.go
@@ -0,0 +1,51 @@
+package ingester
+
+import (
+	"context"
+	"fmt"
+
+	"epimetheus/internal/metrics"
+
+	"github.com/prometheus/client_golang/prometheus/push"
+)
+
+// PushgatewayIngester handles realtime metric ingestion via Pushgateway.
+// Note: Pushgateway does not preserve custom timestamps - all metrics are
+// timestamped with the current time when pushed.
+type PushgatewayIngester struct{}
+
+// NewPushgatewayIngester creates a new Pushgateway ingester.
+func NewPushgatewayIngester() PushgatewayIngester {
+	return PushgatewayIngester{}
+}
+
+// Ingest pushes metrics to Pushgateway.
+// The samples parameter is currently ignored because Pushgateway doesn't support
+// custom metric values from samples - it uses registered Prometheus collectors.
+// This ingests generated metrics using the provided collectors.
+func (i PushgatewayIngester) Ingest(ctx context.Context, collectors metrics.Collectors, url, jobName string) error {
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+
+	// Generate random metric values
+	collectors.Simulate()
+
+	// Create pusher with all collectors
+	pusher := push.New(url, jobName).
+		Collector(collectors.RequestsTotal).
+		Collector(collectors.ActiveConnections).
+		Collector(collectors.TemperatureCelsius).
+		Collector(collectors.RequestDuration).
+		Collector(collectors.JobsProcessed).
+		Grouping("instance", "example-app")
+
+	// Push metrics to Pushgateway
+	if err := pusher.Push(); err != nil {
+		return fmt.Errorf("failed to push to pushgateway: %w", err)
+	}
+
+	return nil
+}
diff --git a/internal/ingester/pushgateway_test.go b/internal/ingester/pushgateway_test.go
new file mode 100644
index 0000000..55fc213
--- /dev/null
+++ b/internal/ingester/pushgateway_test.go
@@ -0,0 +1,28 @@
+package ingester
+
+import (
+	"testing"
+
+	"epimetheus/internal/metrics"
+)
+
+func TestNewPushgatewayIngester(t *testing.T) {
+	ingester := NewPushgatewayIngester()
+
+	// Verify the ingester was created (value type, so no nil check needed)
+	_ = ingester
+}
+
+func TestPushgatewayIngester_Type(t *testing.T) {
+	// Test that we can create and use the ingester
+	collectors := metrics.NewCollectors()
+	ingester := NewPushgatewayIngester()
+
+	// The ingester should work with collectors
+	if collectors.RequestsTotal == nil {
+		t.Error("Collectors not initialized properly")
+	}
+
+	// Verify ingester is the correct type
+	_ = ingester
+}
diff --git a/internal/ingester/remotewrite.go b/internal/ingester/remotewrite.go
new file mode 100644
index 0000000..b88b7fa
--- /dev/null
+++ b/internal/ingester/remotewrite.go
@@ -0,0 +1,455 @@
+package ingester
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"log"
+	"math/rand"
+	"net/http"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	"epimetheus/internal/metrics"
+
+	"github.com/golang/snappy"
+	"github.com/prometheus/prometheus/prompb"
+)
+
+const (
+	requestTimeout = 10 * time.Second
+	backfillDelay  = 100 * time.Millisecond
+	// BatchSize defines the maximum number of samples per Remote Write request
+	// Prometheus has a 32MB limit, so we keep batches small to avoid rejection
+	BatchSize = 5000
+	// NumWorkers defines the number of concurrent goroutines for batch processing
+	// Higher values increase throughput but may overwhelm Prometheus
+	NumWorkers = 10
+)
+
+// Buffer pool for reusing buffers across requests (reduces GC pressure)
+var bufferPool = sync.Pool{
+	New: func() interface{} {
+		return new(bytes.Buffer)
+	},
+}
+
+// Preallocated buffer pool for protobuf marshaling
+var protoBufferPool = sync.Pool{
+	New: func() interface{} {
+		// Preallocate ~500KB buffer (typical batch size)
+		buf := make([]byte, 0, 512*1024)
+		return &buf
+	},
+}
+
+// TimeSeries object pool for reuse
+var timeSeriesPool = sync.Pool{
+	New: func() interface{} {
+		return &prompb.TimeSeries{
+			Labels:  make([]prompb.Label, 0, 10), // Typical label count
+			Samples: make([]prompb.Sample, 0, 1),
+		}
+	},
+}
+
+// Label slice pool for reuse
+var labelSlicePool = sync.Pool{
+	New: func() interface{} {
+		labels := make([]prompb.Label, 0, 10)
+		return &labels
+	},
+}
+
+// RemoteWriteIngester handles historic metric ingestion via Prometheus Remote Write API.
+// This ingester preserves custom timestamps, making it suitable for importing historic data.
+type RemoteWriteIngester struct {
+	client *http.Client
+}
+
+// NewRemoteWriteIngester creates a new Remote Write ingester with optimized HTTP client.
+func NewRemoteWriteIngester() RemoteWriteIngester {
+	// Optimized HTTP transport with connection pooling
+	transport := &http.Transport{
+		MaxIdleConns:        100,              // Global connection pool
+		MaxIdleConnsPerHost: NumWorkers,       // Match worker count
+		MaxConnsPerHost:     NumWorkers,       // Limit concurrent connections per host
+		IdleConnTimeout:     90 * time.Second, // Keep connections alive longer
+		DisableKeepAlives:   false,            // CRITICAL: enable keep-alive
+		DisableCompression:  true,             // We compress manually with Snappy
+		ForceAttemptHTTP2:   true,             // Use HTTP/2 if available
+		WriteBufferSize:     64 * 1024,        // 64KB write buffer
+		ReadBufferSize:      64 * 1024,        // 64KB read buffer
+	}
+
+	return RemoteWriteIngester{
+		client: &http.Client{
+			Timeout:   requestTimeout,
+			Transport: transport,
+		},
+	}
+}
+
+// Ingest sends samples to Prometheus via Remote Write API with batching and concurrency.
+// Large datasets are automatically split into batches and processed by a worker pool
+// to avoid exceeding Prometheus's 32MB limit and maximize throughput.
+func (i RemoteWriteIngester) Ingest(ctx context.Context, samples []metrics.Sample, url string) error {
+	if len(samples) == 0 {
+		return fmt.Errorf("no samples to ingest")
+	}
+
+	select {
+	case <-ctx.Done():
+		return ctx.Err()
+	default:
+	}
+
+	// Split samples into batches
+	batches := chunkSamples(samples, BatchSize)
+	totalBatches := len(batches)
+
+	log.Printf("Splitting %d samples into %d batches (batch size: %d)", len(samples), totalBatches, BatchSize)
+	log.Printf("Processing batches with %d concurrent workers", NumWorkers)
+
+	// Counters for tracking progress (atomic for thread safety)
+	var successCount int32
+	var errorCount int32
+	var processedCount int32
+
+	// Worker pool pattern
+	batchChan := make(chan batchJob, totalBatches)
+	errorsChan := make(chan error, totalBatches)
+	var wg sync.WaitGroup
+
+	// Start workers
+	for w := 0; w < NumWorkers; w++ {
+		wg.Add(1)
+		go func(workerID int) {
+			defer wg.Done()
+
+			for job := range batchChan {
+				// Check context before processing
+				select {
+				case <-ctx.Done():
+					errorsChan <- fmt.Errorf("worker %d cancelled", workerID)
+					return
+				default:
+				}
+
+				// Convert and send batch
+				timeSeries := convertSamplesToTimeSeries(job.batch)
+				writeRequest := &prompb.WriteRequest{Timeseries: timeSeries}
+
+				if err := i.sendWriteRequest(ctx, url, writeRequest); err != nil {
+					atomic.AddInt32(&errorCount, 1)
+					errorsChan <- fmt.Errorf("batch %d: %w", job.index, err)
+				} else {
+					atomic.AddInt32(&successCount, 1)
+				}
+
+				// Update progress
+				processed := atomic.AddInt32(&processedCount, 1)
+				if processed%10 == 0 || int(processed) == totalBatches {
+					progress := float64(processed) / float64(totalBatches) * 100
+					log.Printf("Progress: %.1f%% (%d/%d batches, %d success, %d errors)",
+						progress, processed, totalBatches, atomic.LoadInt32(&successCount), atomic.LoadInt32(&errorCount))
+				}
+			}
+		}(w)
+	}
+
+	// Send batches to workers
+	for idx, batch := range batches {
+		batchChan <- batchJob{index: idx + 1, batch: batch}
+	}
+	close(batchChan)
+
+	// Wait for all workers to finish
+	wg.Wait()
+	close(errorsChan)
+
+	// Collect errors
+	var firstError error
+	errorList := make([]error, 0)
+	for err := range errorsChan {
+		errorList = append(errorList, err)
+		if firstError == nil {
+			firstError = err
+		}
+	}
+
+	finalSuccess := int(atomic.LoadInt32(&successCount))
+	finalErrors := int(atomic.LoadInt32(&errorCount))
+
+	log.Printf("Batch ingestion complete: %d successful, %d errors", finalSuccess, finalErrors)
+
+	if finalErrors > 0 {
+		// Log first few errors as examples
+		numToLog := 5
+		if len(errorList) < numToLog {
+			numToLog = len(errorList)
+		}
+		log.Printf("Sample errors (showing %d of %d):", numToLog, finalErrors)
+		for i := 0; i < numToLog; i++ {
+			log.Printf("  - %v", errorList[i])
+		}
+		return fmt.Errorf("completed with %d/%d batches failed", finalErrors, totalBatches)
+	}
+
+	return nil
+}
+
+// batchJob represents a batch to be processed by a worker.
+type batchJob struct {
+	index int
+	batch []metrics.Sample
+}
+
+// chunkSamples splits samples into batches of the specified size.
+func chunkSamples(samples []metrics.Sample, batchSize int) [][]metrics.Sample {
+	var batches [][]metrics.Sample
+
+	for i := 0; i < len(samples); i += batchSize {
+		end := i + batchSize
+		if end > len(samples) {
+			end = len(samples)
+		}
+		batches = append(batches, samples[i:end])
+	}
+
+	return batches
+}
+
+// IngestHistoric generates and ingests historic metrics for a specific time in the past.
+func (i RemoteWriteIngester) IngestHistoric(ctx context.Context, url string, hoursAgo int) error {
+	timestamp := time.Now().Add(-time.Duration(hoursAgo) * time.Hour)
+	timeSeries := generateHistoricTimeSeries(timestamp)
+	writeRequest := &prompb.Wri