package main

// Cloudflare does not forward container stdout to observability, so syncd
// ships its own output and git-forge's to the worker. Drops rather than
// blocks, and never logs through log itself.

import (
	"bytes"
	"fmt"
	"net/http"
	"os"
	"strings"
	"sync/atomic"
	"time"
)

const (
	shipQueue   = 256
	shipEvery   = 500 * time.Millisecond
	shipTimeout = 3 * time.Second
	// The bound on getting run's attention; the flush itself is waited out,
	// since those lines say how the container stopped.
	shipWait = 5 * time.Second
)

type shipper struct {
	url     string
	lines   chan string
	flush   chan chan struct{}
	dropped atomic.Int64 // never queued, because the queue was full
	unsent  atomic.Int64 // queued, but the send failed
}

// newShipper tees to stdout and to the worker, or to stdout alone with no
// endpoint.
func newShipper(base string) *shipper {
	s := &shipper{
		url:   base,
		lines: make(chan string, shipQueue),
		flush: make(chan chan struct{}),
	}
	if base != "" {
		go s.run()
	}
	return s
}

// close sends what is queued and waits: the last lines say how the container
// shut down.
func (s *shipper) close() {
	if s.url == "" {
		return
	}
	ack := make(chan struct{})
	select {
	case s.flush <- ack:
		<-ack
	case <-time.After(shipWait):
	}
}

// Write always reports success: the caller is a logger.
func (s *shipper) Write(p []byte) (int, error) {
	_, _ = os.Stdout.Write(p)
	if s.url == "" {
		return len(p), nil
	}
	for _, line := range strings.Split(string(bytes.TrimRight(p, "\n")), "\n") {
		if line == "" {
			continue
		}
		select {
		case s.lines <- line:
		default:
			// Dropped rather than stalling a request; counted, so the gap is
			// reported.
			s.dropped.Add(1)
		}
	}
	return len(p), nil
}

func (s *shipper) run() {
	client := &http.Client{Timeout: shipTimeout}
	t := time.NewTicker(shipEvery)
	defer t.Stop()

	var batch []string
	send := func() {
		lines := int64(len(batch))
		dropped, unsent := s.dropped.Swap(0), s.unsent.Swap(0)
		if dropped > 0 {
			batch = append(batch, fmt.Sprintf("syncd: dropped %d log lines, the queue was full", dropped))
		}
		if unsent > 0 {
			batch = append(batch, fmt.Sprintf("syncd: %d log lines did not reach the worker", unsent))
		}
		if len(batch) == 0 {
			return
		}
		body := strings.Join(batch, "\n")
		batch = batch[:0]
		// The status too: a 500 loses the batch as a refused connection does.
		landed := false
		if resp, err := client.Post(s.url, "text/plain", strings.NewReader(body)); err == nil {
			landed = resp.StatusCode/100 == 2
			resp.Body.Close()
		}
		if !landed {
			// Put back, so the next send that works reports one total.
			s.dropped.Add(dropped)
			s.unsent.Add(unsent + lines)
		}
	}

	for {
		select {
		case line := <-s.lines:
			batch = append(batch, line)
			if len(batch) >= shipQueue/2 {
				send()
			}
		case <-t.C:
			send()
		case ack := <-s.flush:
			for drained := false; !drained; {
				select {
				case line := <-s.lines:
					batch = append(batch, line)
				default:
					drained = true
				}
			}
			send()
			close(ack)
		}
	}
}