// Package breaker is a small circuit breaker: after N failures in a row it stops calling a dependency for a
// cool-down period and fails fast, then lets one trial call through (half-open) to see if it has recovered.
package breaker

import (
        "errors"
        "sync"
        "time"
)

// ErrOpen is returned while the circuit is open: the call was not attempted.
var ErrOpen = errors.New("circuit open")

// State of the breaker.
type State string

const (
        Closed   State = "closed"
        Open     State = "open"
        HalfOpen State = "half-open"
)

// Breaker counts consecutive failures. Now is replaceable so tests can move time instead of sleeping.
type Breaker struct {
        mu        sync.Mutex
        threshold int
        cooldown  time.Duration
        failures  int
        openedAt  time.Time
        state     State
        trial     bool
        Now       func() time.Time
}

// New creates a breaker that opens after threshold consecutive failures and stays open for cooldown.
func New(threshold int, cooldown time.Duration) *Breaker {
        return &Breaker{threshold: threshold, cooldown: cooldown, state: Closed, Now: time.Now}
}

// State reports the current state.
func (b *Breaker) State() State {
        b.mu.Lock()
        defer b.mu.Unlock()
        b.advance()
        return b.state
}

func (b *Breaker) advance() {
        if b.state == Open && b.Now().Sub(b.openedAt) >= b.cooldown {
                b.state = HalfOpen
                b.trial = false
        }
}

// Do runs fn unless the circuit is open. In half-open state only one trial call runs at a time.
func (b *Breaker) Do(fn func() error) error {
        b.mu.Lock()
        b.advance()
        switch {
        case b.state == Open:
                b.mu.Unlock()
                return ErrOpen
        case b.state == HalfOpen && b.trial:
                b.mu.Unlock()
                return ErrOpen
        case b.state == HalfOpen:
                b.trial = true
        }
        b.mu.Unlock()

        err := fn()

        b.mu.Lock()
        defer b.mu.Unlock()
        if err == nil {
                b.failures = 0
                b.state = Closed
                b.trial = false
                return nil
        }
        b.failures++
        if b.state == HalfOpen || b.failures >= b.threshold {
                b.state = Open
                b.openedAt = b.Now()
                b.trial = false
        }
        return err
}

// Package chaos is a fault-injecting reverse proxy. It sits between two services and, per request, can add
// latency, answer 500, cut the connection, or swallow the request (black hole) — the faults a real network gives.
//
// Faults are set from the test through Set, so one scenario can switch them on and off mid-flight.
package chaos

import (
        "io"
        "net/http"
        "net/http/httputil"
        "net/url"
        "sync"
        "sync/atomic"
        "time"
)

// Fault describes what the proxy does to the next requests.
type Fault struct {
        Latency   time.Duration // added before forwarding
        SlowFor   int           // if > 0, latency applies only to the next N requests
        Error500  int           // answer the next N requests with 500 without forwarding
        Reset     int           // forward the next N requests, then drop the connection before the response
        BlackHole bool          // never answer (until the client gives up)
        Down      bool          // refuse every request with a connection drop, as if the service were gone
}

// Proxy is the fault injector.
type Proxy struct {
        mu      sync.Mutex
        fault   Fault
        target  *httputil.ReverseProxy
        Seen    atomic.Int64 // requests that reached the proxy
        Dropped atomic.Int64 // requests answered by a fault instead of the target
}

// New creates a proxy in front of target.
func New(target string) *Proxy {
        u, err := url.Parse(target)
        if err != nil {
                panic(err)
        }
        return &Proxy{target: httputil.NewSingleHostReverseProxy(u)}
}

// Set replaces the current fault.
func (p *Proxy) Set(f Fault) {
        p.mu.Lock()
        p.fault = f
        p.mu.Unlock()
}

// take decides the fault for one request and consumes counters.
func (p *Proxy) take() (latency time.Duration, err500, reset, hole, down bool) {
        p.mu.Lock()
        defer p.mu.Unlock()
        f := &p.fault
        latency, hole, down = f.Latency, f.BlackHole, f.Down
        if f.SlowFor > 0 {
                f.SlowFor--
                if f.SlowFor == 0 {
                        f.Latency = 0
                }
        }
        if f.Error500 > 0 {
                f.Error500--
                err500 = true
        } else if f.Reset > 0 {
                f.Reset--
                reset = true
        }
        return
}

func hijackClose(w http.ResponseWriter) {
        if hj, ok := w.(http.Hijacker); ok {
                if conn, _, err := hj.Hijack(); err == nil {
                        _ = conn.Close()
                        return
                }
        }
        panic(http.ErrAbortHandler)
}

// ServeHTTP applies the fault, then forwards.
func (p *Proxy) ServeHTTP(w http.ResponseWriter, r *http.Request) {
        p.Seen.Add(1)
        latency, err500, reset, hole, down := p.take()
        if down {
                p.Dropped.Add(1)
                hijackClose(w)
                return
        }
        if hole {
                p.Dropped.Add(1)
                // Read the body first: the server only notices the client hanging up once the body is consumed.
                _, _ = io.Copy(io.Discard, r.Body)
                select {
                case <-r.Context().Done():
                case <-time.After(time.Minute):
                }
                return
        }
        if latency > 0 {
                select {
                case <-time.After(latency):
                case <-r.Context().Done():
                        // The client gave up. The request is still forwarded, the way a slow network still delivers it:
                        // that is what makes a retry after a timeout dangerous without an idempotency key.
                }
        }
        if err500 {
                p.Dropped.Add(1)
                w.Header().Set("Content-Type", "application/json")
                w.WriteHeader(http.StatusInternalServerError)
                _, _ = w.Write([]byte(`{"error":"injected"}`))
                return
        }
        if reset {
                // Forward for real (the reservation happens), then drop the connection instead of the response.
                p.Dropped.Add(1)
                rec := &discard{header: http.Header{}}
                p.target.ServeHTTP(rec, r.Clone(contextWithoutCancel(r)))
                hijackClose(w)
                return
        }
        p.target.ServeHTTP(w, r.Clone(contextWithoutCancel(r)))
}

type discard struct{ header http.Header }

func (d *discard) Header() http.Header         { return d.header }
func (d *discard) Write(b []byte) (int, error) { return len(b), nil }
func (d *discard) WriteHeader(int)             {}

package chaos

import (
        "context"
        "net/http"
)

func contextWithoutCancel(r *http.Request) context.Context { return context.WithoutCancel(r.Context()) }

// Package inventory is a stock service: it holds how many units of each SKU are left and reserves them.
//
// A reservation carries an Idempotency-Key. Sending the same key again returns the first result instead of
// reserving twice, which is what makes it safe for the orders service to retry after a timeout.
package inventory

import (
        "bytes"
        "encoding/json"
        "errors"
        "io"
        "net/http"
        "strings"
        "sync"
)

// Reservation is the body of POST /reservations.
type Reservation struct {
        SKU      string `json:"sku"`
        Quantity int    `json:"quantity"`
}

// Validate checks a reservation before it touches the stock.
func (r Reservation) Validate() error {
        switch {
        case strings.TrimSpace(r.SKU) == "" || len(r.SKU) > 32:
                return errors.New("sku must be 1-32 characters")
        case r.Quantity < 1 || r.Quantity > 100:
                return errors.New("quantity must be 1-100")
        }
        return nil
}

// ParseReservation decodes and validates a request body. Unknown fields and trailing data are rejected.
func ParseReservation(body []byte) (Reservation, error) {
        var r Reservation
        dec := json.NewDecoder(bytes.NewReader(body))
        dec.DisallowUnknownFields()
        if err := dec.Decode(&r); err != nil {
                return Reservation{}, errors.New("invalid JSON: " + err.Error())
        }
        if dec.More() {
                return Reservation{}, errors.New("invalid JSON: trailing data")
        }
        return r, r.Validate()
}

type result struct {
        status int
        body   any
}

// Store is the inventory: stock per SKU and the results of reservations already made, by idempotency key.
type Store struct {
        mu       sync.Mutex
        stock    map[string]int
        seen     map[string]result
        Reserved int // total units reserved, for tests that count side effects
        Requests int // reservation requests received, including repeats
}

// NewStore creates a store with the given stock.
func NewStore(stock map[string]int) *Store {
        s := &Store{stock: map[string]int{}, seen: map[string]result{}}
        for k, v := range stock {
                s.stock[k] = v
        }
        return s
}

// Stock returns the units left for a SKU and whether the SKU exists.
func (s *Store) Stock(sku string) (int, bool) {
        s.mu.Lock()
        defer s.mu.Unlock()
        n, ok := s.stock[sku]
        return n, ok
}

// Reserve takes units out of stock, at most once per idempotency key.
func (s *Store) Reserve(key string, r Reservation) (int, any) {
        s.mu.Lock()
        defer s.mu.Unlock()
        s.Requests++
        if prev, ok := s.seen[key]; ok && key != "" {
                return prev.status, prev.body
        }
        var res result
        n, ok := s.stock[r.SKU]
        switch {
        case !ok:
                res = result{http.StatusNotFound, map[string]string{"error": "unknown sku"}}
        case n < r.Quantity:
                res = result{http.StatusConflict, map[string]any{"error": "not enough stock", "available": n}}
        default:
                s.stock[r.SKU] = n - r.Quantity
                s.Reserved += r.Quantity
                res = result{http.StatusCreated, map[string]any{"sku": r.SKU, "quantity": r.Quantity, "left": n - r.Quantity}}
        }
        if key != "" {
                s.seen[key] = res
        }
        return res.status, res.body
}

// Handler serves GET /items/{sku} and POST /reservations.
func (s *Store) Handler() http.Handler {
        mux := http.NewServeMux()
        mux.HandleFunc("GET /items/{sku}", func(w http.ResponseWriter, r *http.Request) {
                n, ok := s.Stock(r.PathValue("sku"))
                if !ok {
                        writeJSON(w, http.StatusNotFound, map[string]string{"error": "unknown sku"})
                        return
                }
                writeJSON(w, http.StatusOK, map[string]any{"sku": r.PathValue("sku"), "stock": n})
        })
        mux.HandleFunc("POST /reservations", func(w http.ResponseWriter, r *http.Request) {
                body, _ := io.ReadAll(io.LimitReader(r.Body, 4097))
                if len(body) > 4096 {
                        writeJSON(w, http.StatusRequestEntityTooLarge, map[string]string{"error": "body too large"})
                        return
                }
                res, err := ParseReservation(body)
                if err != nil {
                        writeJSON(w, http.StatusBadRequest, map[string]string{"error": err.Error()})
                        return
                }
                status, out := s.Reserve(r.Header.Get("Idempotency-Key"), res)
                writeJSON(w, status, out)
        })
        return mux
}

func writeJSON(w http.ResponseWriter, status int, v any) {
        w.Header().Set("Content-Type", "application/json")
        w.WriteHeader(status)
        _ = json.NewEncoder(w).Encode(v)
}

// Package orders is the front service: POST /orders reserves stock in the inventory service and records the order.
//
// The call to inventory is where things go wrong in a distributed system, so it is wrapped in three defences:
//   - a timeout per attempt, so a slow inventory cannot hang the order;
//   - retries with backoff for timeouts, resets and 5xx, all with the same Idempotency-Key, so a retry after a lost
//     response never reserves twice;
//   - a circuit breaker, so when inventory is down orders fail fast with 503 instead of piling up.
package orders

import (
        "bytes"
        "context"
        "encoding/json"
        "errors"
        "fmt"
        "io"
        "net/http"
        "strconv"
        "sync"
        "sync/atomic"
        "time"

        "github.com/NailKarimov/go-testing/breaker"
)

// Config tunes the client side of the inventory call.
type Config struct {
        InventoryURL string
        Timeout      time.Duration // per attempt
        Attempts     int           // first try + retries
        Backoff      time.Duration // doubled after each retry
        Breaker      *breaker.Breaker
        Client       *http.Client
}

// Order is what the service stores and returns.
type Order struct {
        ID       int    `json:"id"`
        SKU      string `json:"sku"`
        Quantity int    `json:"quantity"`
        Status   string `json:"status"`
}

// Service is the orders service.
type Service struct {
        cfg    Config
        mu     sync.Mutex
        orders []Order
        nextID atomic.Int64
        // Attempts counts calls made to inventory, for tests that check retry behaviour.
        Attempts atomic.Int64
}

// New creates the service with defaults for anything left empty.
func New(cfg Config) *Service {
        if cfg.Timeout == 0 {
                cfg.Timeout = 300 * time.Millisecond
        }
        if cfg.Attempts == 0 {
                cfg.Attempts = 3
        }
        if cfg.Backoff == 0 {
                cfg.Backoff = 20 * time.Millisecond
        }
        if cfg.Breaker == nil {
                cfg.Breaker = breaker.New(5, 2*time.Second)
        }
        if cfg.Client == nil {
                cfg.Client = &http.Client{}
        }
        return &Service{cfg: cfg}
}

// errRetryable marks failures worth another attempt: timeouts, resets, 5xx.
type errRetryable struct{ cause error }

func (e errRetryable) Error() string { return e.cause.Error() }

// reply is a final answer from inventory: 201 created, 404 or 409 are business answers, not failures.
type reply struct {
        status int
        body   map[string]any
}

func (s *Service) reserveOnce(ctx context.Context, key string, payload []byte) (reply, error) {
        s.Attempts.Add(1)
        ctx, cancel := context.WithTimeout(ctx, s.cfg.Timeout)
        defer cancel()
        req, err := http.NewRequestWithContext(ctx, http.MethodPost, s.cfg.InventoryURL+"/reservations", bytes.NewReader(payload))
        if err != nil {
                return reply{}, err
        }
        req.Header.Set("Content-Type", "application/json")
        req.Header.Set("Idempotency-Key", key)
        resp, err := s.cfg.Client.Do(req)
        if err != nil {
                return reply{}, errRetryable{err}
        }
        defer func() { _ = resp.Body.Close() }()
        raw, err := io.ReadAll(io.LimitReader(resp.Body, 1<<16))
        if err != nil {
                return reply{}, errRetryable{err} // connection dropped mid-body
        }
        if resp.StatusCode >= 500 {
                return reply{}, errRetryable{fmt.Errorf("inventory answered %d", resp.StatusCode)}
        }
        var body map[string]any
        if err := json.Unmarshal(raw, &body); err != nil {
                return reply{}, errRetryable{fmt.Errorf("inventory sent invalid JSON: %w", err)}
        }
        return reply{resp.StatusCode, body}, nil
}

// reserve calls inventory through the breaker with retries. Every attempt carries the same idempotency key.
func (s *Service) reserve(ctx context.Context, key string, payload []byte) (reply, error) {
        var out reply
        err := s.cfg.Breaker.Do(func() error {
                wait := s.cfg.Backoff
                var last error
                for i := 0; i < s.cfg.Attempts; i++ {
                        if i > 0 {
                                select {
                                case <-time.After(wait):
                                case <-ctx.Done():
                                        return ctx.Err()
                                }
                                wait *= 2
                        }
                        r, err := s.reserveOnce(ctx, key, payload)
                        if err == nil {
                                out = r
                                return nil
                        }
                        last = err
                        var re errRetryable
                        if !errors.As(err, &re) {
                                return err
                        }
                }
                return last
        })
        return out, err
}

// Handler serves POST /orders and GET /orders.
func (s *Service) Handler() http.Handler {
        mux := http.NewServeMux()
        mux.HandleFunc("POST /orders", func(w http.ResponseWriter, r *http.Request) {
                var in struct {
                        SKU      string `json:"sku"`
                        Quantity int    `json:"quantity"`
                }
                dec := json.NewDecoder(io.LimitReader(r.Body, 4096))
                dec.DisallowUnknownFields()
                if err := dec.Decode(&in); err != nil || in.SKU == "" || in.Quantity < 1 || in.Quantity > 100 {
                        writeJSON(w, http.StatusBadRequest, map[string]string{"error": "sku and quantity 1-100 are required"})
                        return
                }
                id := int(s.nextID.Add(1))
                payload, _ := json.Marshal(in)
                res, err := s.reserve(r.Context(), "order-"+strconv.Itoa(id), payload)
                switch {
                case errors.Is(err, breaker.ErrOpen):
                        w.Header().Set("Retry-After", "2")
                        writeJSON(w, http.StatusServiceUnavailable, map[string]string{"error": "inventory unavailable (circuit open)"})
                        return
                case err != nil:
                        writeJSON(w, http.StatusBadGateway, map[string]string{"error": "inventory failed: " + err.Error()})
                        return
                case res.status == http.StatusNotFound:
                        writeJSON(w, http.StatusUnprocessableEntity, map[string]string{"error": "unknown sku"})
                        return
                case res.status == http.StatusConflict:
                        writeJSON(w, http.StatusConflict, map[string]any{"error": "out of stock", "available": res.body["available"]})
                        return
                case res.status != http.StatusCreated:
                        writeJSON(w, http.StatusBadGateway, map[string]string{"error": fmt.Sprintf("unexpected inventory status %d", res.status)})
                        return
                }
                o := Order{ID: id, SKU: in.SKU, Quantity: in.Quantity, Status: "confirmed"}
                s.mu.Lock()
                s.orders = append(s.orders, o)
                s.mu.Unlock()
                writeJSON(w, http.StatusCreated, o)
        })
        mux.HandleFunc("GET /orders", func(w http.ResponseWriter, _ *http.Request) {
                s.mu.Lock()
                defer s.mu.Unlock()
                writeJSON(w, http.StatusOK, append([]Order{}, s.orders...))
        })
        mux.HandleFunc("GET /health", func(w http.ResponseWriter, _ *http.Request) {
                writeJSON(w, http.StatusOK, map[string]string{"breaker": string(s.cfg.Breaker.State())})
        })
        return mux
}

func writeJSON(w http.ResponseWriter, status int, v any) {
        w.Header().Set("Content-Type", "application/json")
        w.WriteHeader(status)
        _ = json.NewEncoder(w).Encode(v)
}