package main

// logshipper - tails the pinball machine's gameplay logs and pushes them to Loki.
//
// Tails /dump/mode.log and /dump/switch.log, batches their lines, and POSTs them to
// Loki's /loki/api/v1/push. Per-file read offsets are saved only after a successful
// push, so a down Loki means retry-then-catch-up, never loss. Configuration is env:
//
//	LOKI_URL        push target        (default https://loki.k8s-home.lqx.net)
//	SHIP_FILES      comma list         (default /dump/mode.log,/dump/switch.log)
//	SHIP_STATE      offset state file  (default /dump/.logshipper.state)
//	MACHINE         the machine label  (default this host's name)
//	SHIP_FROM_START 1 = ship from each file's start on first run (default: from its end)
//	SHIP_INSECURE   1 = skip TLS verification (for a Loki without a trusted cert)
//	SHIP_EVERY      poll interval      (default 2s)
//
// Runs under /etc/init.d/log_shipper; its stderr goes to the syslog.

import (
	"bytes"
	"crypto/tls"
	"fmt"
	"io"
	"log"
	"net/http"
	"os"
	"os/signal"
	"strings"
	"syscall"
	"time"
)

var (
	lokiURL   = envOr("LOKI_URL", "https://loki.k8s-home.lqx.net")
	files     = strings.Split(envOr("SHIP_FILES", "/dump/mode.log,/dump/switch.log"), ",")
	statePath = envOr("SHIP_STATE", "/dump/.logshipper.state")
	machine   = envOr("MACHINE", hostname())
	fromStart = os.Getenv("SHIP_FROM_START") == "1"
	insecure  = os.Getenv("SHIP_INSECURE") == "1"
	every     = 2 * time.Second
	maxPush   = 3 << 20 // under Loki's 4 MiB push limit
)

func envOr(key, fallback string) string {
	if v := os.Getenv(key); v != "" {
		return v
	}
	return fallback
}

func hostname() string {
	if h, err := os.Hostname(); err == nil && h != "" {
		return h
	}
	return "stern-spike2"
}

// readNewLines returns the complete lines added to path since offset, and the offset to
// save once they are pushed (never past the last newline). A file truncated under us
// (": > mode.log" is safe housekeeping) restarts from 0.
func readNewLines(path string, offset int64) (lines []string, sent int64, err error) {
	f, err := os.Open(path)
	if err != nil {
		return nil, offset, err
	}
	defer f.Close()
	fi, err := f.Stat()
	if err != nil {
		return nil, offset, err
	}
	resume := offset
	if fi.Size() < offset {
		resume = 0
	}
	if _, err = f.Seek(resume, io.SeekStart); err != nil {
		return nil, offset, err
	}
	buf, err := io.ReadAll(f)
	if err != nil {
		return nil, resume, err
	}
	s := string(buf)
	if cut := strings.LastIndexByte(s, '\n'); cut >= 0 {
		lines = strings.Split(strings.TrimSuffix(s[:cut+1], "\n"), "\n")
		sent = resume + int64(cut) + 1
	} else {
		sent = resume // no complete line yet
	}
	return lines, sent, nil
}

func push(payload []byte) error {
	return post("/loki/api/v1/push", payload)
}

// pushFlush asks Loki to write the ingester's chunks to the store now.
func pushFlush() error {
	return post("/flush", nil)
}

func post(path string, payload []byte) error {
	transport := http.DefaultTransport.(*http.Transport).Clone()
	if insecure {
		transport.TLSClientConfig = &tls.Config{InsecureSkipVerify: true}
	}
	client := &http.Client{Timeout: 30 * time.Second, Transport: transport}
	var body io.Reader
	if payload != nil {
		body = bytes.NewReader(payload)
	}
	resp, err := client.Post(lokiURL+path, "application/json", body)
	if err != nil {
		return err
	}
	defer resp.Body.Close()
	read, _ := io.ReadAll(io.LimitReader(resp.Body, 300))
	if resp.StatusCode != http.StatusNoContent && resp.StatusCode != http.StatusOK {
		return fmt.Errorf("loki said %s: %s", resp.Status, strings.TrimSpace(string(read)))
	}
	return nil
}

type shipFile struct {
	path   string
	source string
	game   string
}

func main() {
	log.SetFlags(log.LstdFlags | log.LUTC)
	if v := os.Getenv("SHIP_EVERY"); v != "" {
		if d, err := time.ParseDuration(v); err == nil && d > 0 {
			every = d
		}
	}
	st, err := loadState(statePath)
	if err != nil {
		log.Printf("state file unreadable (%v) - starting fresh", err)
		st = state{}
	}
	var ship []shipFile
	for _, path := range files {
		path = strings.TrimSpace(path)
		if path == "" {
			continue
		}
		ship = append(ship, shipFile{path: path, source: baseName(path)})
	}
	log.Printf("shipping %s to %s (labels: job=pinball machine=%s, every %s)",
		strings.Join(files, ", "), lokiURL, machine, every)

	sig := make(chan os.Signal, 1)
	signal.Notify(sig, syscall.SIGINT, syscall.SIGTERM)
	tick := time.Tick(every)
	for {
		for i := range ship {
			f := &ship[i]
			offset, known := st[baseName(f.path)]
			if !known && !fromStart {
				// first run: start from the file's end; run once with SHIP_FROM_START=1
				// to ship the history already on disk
				if fi, err := os.Stat(f.path); err == nil {
					offset = fi.Size()
				}
			}
			lines, sent, err := readNewLines(f.path, offset)
			if err != nil {
				if !os.IsNotExist(err) {
					log.Printf("%s: %v", f.path, err)
				}
				continue
			}
			if len(lines) == 0 {
				continue
			}
			stamps := assignTimestamps(lines, time.Now())
			var entries []Entry
			for i, line := range lines {
				if g := extractGame(line); g != "" {
					f.game = g
				}
				entries = append(entries, Entry{Source: f.source, Game: f.game, TS: stamps[i], Line: line})
			}
			pushed := true
			for _, payload := range buildPayloads(entries, machine, maxPush) {
				if err := push(payload); err != nil {
					log.Printf("push failed (%v) - %s stays at offset %d", err, f.path, offset)
					pushed = false
					break
				}
			}
			if !pushed {
				continue // the offset is saved only when the whole batch is in Loki
			}
			if needsFlush(entries, time.Now()) {
				// a backfill: the old entries are invisible in the ingester's head until
				// they reach the store (live tails never come through here)
				if err := pushFlush(); err != nil {
					log.Printf("flush failed (%v) - old entries may not be queryable yet", err)
				}
			}
			st[baseName(f.path)] = sent
			if err := saveState(statePath, st); err != nil {
				log.Printf("state save failed: %v", err)
			}
		}
		select {
		case <-sig:
			log.Printf("stopping")
			return
		case <-tick:
		}
	}
}

func baseName(path string) string {
	if i := strings.LastIndexByte(path, '/'); i >= 0 {
		return path[i+1:]
	}
	return path
}