package main

import (
	"bufio"
	"bytes"
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"net"
	"net/http"
	"os"
	"os/exec"
	"path/filepath"
	"regexp"
	"sort"
	"strconv"
	"strings"
	"syscall"
	"time"
)

var npmTrafficPattern = regexp.MustCompile(`^\[([^]]+)\] .*? (\d{3}) - ([A-Z]+) (?:https?|HTTP) (\S+) "([^"]*)" \[Client ([^]]+)\]`)

func parseNPMTraffic(line string) (TrafficEvent, bool) {
	m := npmTrafficPattern.FindStringSubmatch(line)
	if m == nil {
		return parseNPMPlusTraffic(line)
	}
	ip := net.ParseIP(m[6])
	if ip == nil {
		return TrafficEvent{}, false
	}
	ts, err := time.Parse("02/Jan/2006:15:04:05 -0700", m[1])
	if err != nil {
		return TrafficEvent{}, false
	}
	status, _ := strconv.Atoi(m[2])
	return TrafficEvent{Time: ts.Unix(), Kind: "request", IP: ip.String(), Domain: trafficText(m[4], 253), Method: m[3], Path: trafficPath(m[5]), Status: status}, true
}

type trafficCursor struct {
	Offset     int64 `json:"offset"`
	Generation int64 `json:"generation"`
	Seen       int64 `json:"seen"`
}
type trafficAgentState struct {
	BacklogBytes int64                    `json:"backlogBytes"`
	Initialized  bool                     `json:"initialized"`
	StartedAt    int64                    `json:"startedAt"`
	Cursors      map[string]trafficCursor `json:"cursors"`
	Pending      []TrafficEvent           `json:"pending"`
	ParseErrors  uint64                   `json:"parseErrors"`
	CrowdSeen    map[string]int64         `json:"crowdSeen"`
}

func (state *trafficAgentState) readLogs(dirs ...string) (int, error) {
	state.BacklogBytes = 0
	var paths []string
	seenPaths := map[string]bool{}
	for _, dir := range dirs {
		for _, pattern := range []string{"*_access.log*", "access.log*"} {
			matches, err := filepath.Glob(filepath.Join(dir, pattern))
			if err != nil {
				return 0, err
			}
			for _, path := range matches {
				if !seenPaths[path] {
					paths = append(paths, path)
					seenPaths[path] = true
				}
			}
		}
	}
	sort.Strings(paths)
	files := 0
	budget := int64(1 << 20)
	now := time.Now().Unix()
	var firstErr error
	for _, path := range paths {
		if strings.HasSuffix(path, ".gz") {
			continue
		}
		f, err := os.Open(path)
		if err != nil {
			firstErr = err
			continue
		}
		info, err := f.Stat()
		if err != nil || !info.Mode().IsRegular() {
			f.Close()
			continue
		}
		st, ok := info.Sys().(*syscall.Stat_t)
		if !ok {
			f.Close()
			continue
		}
		active := strings.HasSuffix(path, "_access.log") || filepath.Base(path) == "access.log"
		if active {
			files++
		}
		key := fmt.Sprintf("%d:%d", st.Dev, st.Ino)
		cursor, known := state.Cursors[key]
		if !known {
			if !state.Initialized || !active {
				cursor.Offset = info.Size()
			}
			cursor.Generation = now
		}
		cursor.Seen = now
		if info.Size() < cursor.Offset {
			cursor.Offset = 0
			cursor.Generation++
		}
		if len(state.Pending) < 400 && budget > 0 {
			_, err = f.Seek(cursor.Offset, io.SeekStart)
			if err == nil {
				reader := bufio.NewReaderSize(io.LimitReader(f, budget), 64<<10)
				for len(state.Pending) < 400 && budget > 0 {
					line, readErr := reader.ReadString('\n')
					// Keep an unfinished line for the next poll. Bound malformed huge lines.
					if readErr != nil {
						if len(line) >= 64<<10 {
							cursor.Offset += int64(len(line))
							state.ParseErrors++
						}
						break
					}
					offset := cursor.Offset
					cursor.Offset += int64(len(line))
					budget -= int64(len(line))
					if len(line) > 64<<10 {
						state.ParseErrors++
						continue
					}
					e, ok := parseNPMTraffic(line)
					if !ok {
						state.ParseErrors++
						continue
					}
					if e.Time < now-86400 {
						continue
					}
					e.ID = fmt.Sprintf("npm:%s:%d:%d", key, cursor.Generation, offset)
					state.Pending = append(state.Pending, e)
				}
			} else {
				firstErr = err
			}
		}
		if info.Size() > cursor.Offset {
			state.BacklogBytes += info.Size() - cursor.Offset
		}
		state.Cursors[key] = cursor
		f.Close()
	}
	for key, cursor := range state.Cursors {
		if cursor.Seen < now-7*86400 {
			delete(state.Cursors, key)
		}
	}
	if files > 0 {
		state.Initialized = true
	}
	if firstErr != nil {
		return files, errors.New("Не удалось прочитать часть access-логов NPM")
	}
	if files == 0 {
		return files, errors.New("Access-логи NPM не найдены")
	}
	return files, nil
}

type crowdTrafficAlert struct {
	ID        int64  `json:"id"`
	UUID      string `json:"uuid"`
	CreatedAt string `json:"created_at"`
	Scenario  string `json:"scenario"`
	Simulated bool   `json:"simulated"`
	Source    struct {
		IP string `json:"ip"`
	} `json:"source"`
	Events []struct {
		Meta []struct {
			Key   string `json:"key"`
			Value string `json:"value"`
		} `json:"meta"`
	} `json:"events"`
	Decisions []struct {
		ID        int64  `json:"id"`
		Type      string `json:"type"`
		Value     string `json:"value"`
		Simulated bool   `json:"simulated"`
	} `json:"decisions"`
}

func parseCrowdTraffic(data []byte) ([]TrafficEvent, error) {
	var alerts []crowdTrafficAlert
	if err := json.Unmarshal(data, &alerts); err != nil {
		return nil, err
	}
	out := []TrafficEvent{}
	for _, a := range alerts {
		if a.Simulated || net.ParseIP(a.Source.IP) == nil {
			continue
		}
		ts, err := time.Parse(time.RFC3339Nano, a.CreatedAt)
		if err != nil {
			continue
		}
		id := a.UUID
		if id == "" {
			id = fmt.Sprintf("%d:%d", a.ID, ts.Unix())
		}
		// An alert may span multiple domains. Do not invent a one-to-one link.
		domains := map[string]bool{}
		for _, event := range a.Events {
			for _, m := range event.Meta {
				if m.Key == "target_fqdn" {
					domains[m.Value] = true
				}
			}
		}
		domain := ""
		if len(domains) == 1 {
			for d := range domains {
				domain = d
			}
		}
		out = append(out, TrafficEvent{ID: "cs-alert:" + id, Time: ts.Unix(), Kind: "alert", IP: a.Source.IP, Domain: domain, Reason: a.Scenario})
		for _, d := range a.Decisions {
			if !d.Simulated && net.ParseIP(d.Value) != nil {
				out = append(out, TrafficEvent{ID: fmt.Sprintf("cs-decision:%s:%d", id, d.ID), Time: ts.Unix(), Kind: "decision", IP: d.Value, Domain: domain, Reason: d.Type + ": " + a.Scenario})
			}
		}
	}
	return out, nil
}

// cscli uses the existing local machine credentials. No new bouncer key or
// public LAPI port is needed, and only the read-only alerts list is executed.
func collectCrowdTraffic(ctx context.Context, container ...string) ([]TrafficEvent, bool, error) {
	ctx, cancel := context.WithTimeout(ctx, 6*time.Second)
	defer cancel()
	var data []byte
	if len(container) > 0 && container[0] != "" {
		var err error
		data, err = dockerCrowdOutput(ctx, container[0])
		if err != nil {
			return nil, false, errors.New("CrowdSec в Docker недоступен; проверьте контейнер и Docker socket")
		}
	} else {
		cmd := exec.CommandContext(ctx, "cscli", "alerts", "list", "--since", "24h", "--limit", "50", "-o", "json")
		var output limitedTrafficOutput
		cmd.Stdout = &output
		if err := cmd.Run(); err != nil {
			return nil, false, errors.New("cscli недоступен на хосте; контейнер CrowdSec не обнаружен")
		}
		data = output.Bytes()
	}
	events, err := parseCrowdTraffic(data)
	if err != nil {
		return nil, false, errors.New("Не удалось разобрать JSON CrowdSec")
	}
	var raw []json.RawMessage
	json.Unmarshal(data, &raw)
	return events, len(raw) >= 50, nil
}

type limitedTrafficOutput struct{ bytes.Buffer }

func (b *limitedTrafficOutput) Write(p []byte) (int, error) {
	if b.Len()+len(p) > 4<<20 {
		return 0, errors.New("CrowdSec response too large")
	}
	return b.Buffer.Write(p)
}

func saveTrafficAgentState(path string, state trafficAgentState) error {
	data, err := json.Marshal(state)
	if err != nil {
		return err
	}
	f, err := os.CreateTemp(filepath.Dir(path), ".traffic-")
	if err != nil {
		return err
	}
	name := f.Name()
	defer os.Remove(name)
	if err = f.Chmod(0600); err == nil {
		_, err = f.Write(data)
	}
	if err == nil {
		err = f.Sync()
	}
	closeErr := f.Close()
	if err != nil {
		return err
	}
	if closeErr != nil {
		return closeErr
	}
	return os.Rename(name, path)
}
func runAgentTraffic(ctx context.Context, config agentConfig, stateDir string) {
	dir := os.Getenv("DASHBOARD_NPM_LOG_DIR")
	if dir == "off" {
		return
	}

	statePath := filepath.Join(stateDir, "traffic-state.json")
	state := trafficAgentState{StartedAt: time.Now().Unix(), Cursors: map[string]trafficCursor{}, CrowdSeen: map[string]int64{}}
	if data, err := os.ReadFile(statePath); err == nil {
		if json.Unmarshal(data, &state) != nil {
			fmt.Fprintln(os.Stderr, "traffic: invalid checkpoint; collector stopped")
			return
		}
	} else if !os.IsNotExist(err) {
		fmt.Fprintln(os.Stderr, "traffic: checkpoint unreadable")
		return
	}
	if state.Cursors == nil {
		state.Cursors = map[string]trafficCursor{}
	}
	if state.CrowdSeen == nil {
		state.CrowdSeen = map[string]int64{}
	}
	client := &http.Client{Timeout: 10 * time.Second}
	lastCrowd := time.Time{}
	lastDiscovery := time.Time{}
	discovery := trafficDiscovery{}
	crowdAt := int64(0)
	crowdError := ""
	crowdLimited := false
	ticker := time.NewTicker(2 * time.Second)
	defer ticker.Stop()
	for {
		select {
		case <-ctx.Done():
			return
		default:
		}
		if time.Since(lastDiscovery) >= time.Minute {
			discovery = discoverTraffic(ctx)
			lastDiscovery = time.Now()
		}
		files, logErr := state.readLogs(discovery.Dirs...)
		if logErr != nil {
			logErr = fmt.Errorf("%s; проверьте access-логи и LOGROTATE=true у NPM+", logErr)
		}
		if len(state.Pending) < 400 && time.Since(lastCrowd) >= 15*time.Second {
			events, limited, err := collectCrowdTraffic(ctx, discovery.CrowdContainer)
			if discovery.Error != "" {
				if _, localErr := exec.LookPath("cscli"); localErr != nil {
					err = errors.New(discovery.Error)
				}
			}
			crowdLimited = limited
			lastCrowd = time.Now()
			if err != nil {
				crowdError = err.Error()
			} else {
				crowdError = ""
				crowdAt = time.Now().Unix()
				for _, e := range events {
					if _, seen := state.CrowdSeen[e.ID]; !seen && len(state.Pending) < 500 {
						state.Pending = append(state.Pending, e)
						state.CrowdSeen[e.ID] = e.Time
					}
				}
				for id, ts := range state.CrowdSeen {
					if ts < time.Now().Unix()-86400 {
						delete(state.CrowdSeen, id)
					}
				}
			}
		}
		// Persist the pending batch AND offsets before sending. A retry/restart may
		// resend events; the server's (agent,event_id) unique key makes it idempotent.
		if err := saveTrafficAgentState(statePath, state); err != nil {
			fmt.Fprintln(os.Stderr, "traffic: checkpoint write failed; collector stopped")
			return
		}
		batch := TrafficBatch{BacklogBytes: state.BacklogBytes, CrowdSecLimited: crowdLimited, Events: state.Pending, Files: files, ParseErrors: state.ParseErrors, CrowdSecAt: crowdAt, CrowdSecError: crowdError, StartedAt: state.StartedAt}
		if logErr != nil {
			batch.LogError = logErr.Error()
		}
		data, _ := json.Marshal(batch)
		req, err := http.NewRequestWithContext(ctx, "POST", config.Server+"/api/agent/traffic", bytes.NewReader(data))
		if err != nil {
			return
		}
		req.Header.Set("Authorization", "Bearer "+config.ID+"."+config.Secret)
		req.Header.Set("Content-Type", "application/json")
		if res, err := client.Do(req); err == nil {
			io.Copy(io.Discard, io.LimitReader(res.Body, 4096))
			res.Body.Close()
			if res.StatusCode == 200 {
				state.Pending = nil
				if saveTrafficAgentState(statePath, state) != nil {
					return
				}
			}
		}
		select {
		case <-ctx.Done():
			return
		case <-ticker.C:
		}
	}
}

func startAgentTraffic(config agentConfig, stateDir string) func() {
	ctx, cancel := context.WithCancel(context.Background())
	done := make(chan struct{})
	go func() { defer close(done); runAgentTraffic(ctx, config, stateDir) }()
	return func() { cancel(); <-done }
}