package benchimport

import (
	"bufio"
	"context"
	"encoding/csv"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"path/filepath"
	"strconv"
	"strings"
)

// structuredSource is the FileSource backed by a structured roster file, decoded with the standard library
// only. It supports two formats picked by extension: CSV with a header row, and JSONL (one JSON object per
// line). Each well formed row becomes a BenchRecord; a malformed row is skipped and counted, never aborts
// the batch. The structured fields here are the same workforce tokens the resume path produces, so the
// downstream ingest is identical regardless of source.
type structuredSource struct{}

// NewStructuredSource builds a structured roster FileSource.
func NewStructuredSource() FileSource { return structuredSource{} }

// Resumes on a structured source delegates to the resume reader is not meaningful; a structured source
// has no resume directory. It returns an empty result so a caller that picked the structured source by
// mistake gets nothing rather than a panic.
func (structuredSource) Resumes(ctx context.Context, dir string, yield func(rec BenchRecord, file string) error) (BatchResult, error) {
	_ = ctx
	_ = dir
	_ = yield
	return BatchResult{}, nil
}

// Profiles decodes the roster at path and yields a BenchRecord per well formed row. The format is chosen
// by extension: .csv uses the header mapped CSV decoder, .jsonl or .ndjson or .json uses the line
// delimited JSON decoder. The batch identifier passed to yield is the file base name, recorded against
// bench_import_log.
func (s structuredSource) Profiles(ctx context.Context, path string, yield func(rec BenchRecord, batch string) error) (BatchResult, error) {
	f, err := os.Open(path)
	if err != nil {
		return BatchResult{}, err
	}
	defer func() { _ = f.Close() }()

	batch := filepath.Base(path)
	ext := strings.ToLower(filepath.Ext(path))
	switch ext {
	case ".csv":
		return decodeCSV(ctx, f, batch, yield)
	case ".jsonl", ".ndjson", ".json":
		return decodeJSONL(ctx, f, batch, yield)
	default:
		return BatchResult{}, fmt.Errorf("benchimport: unsupported roster extension %q", ext)
	}
}

// decodeCSV reads a header row, maps the standard bench columns by lowercased header name, and yields one
// BenchRecord per data row. A row with the wrong field count or no usable signal is skipped and counted.
// Columns are matched flexibly (name, location, skills, certs, agencies, years, clearance, email, phone,
// external_id) with a few common aliases.
func decodeCSV(ctx context.Context, r io.Reader, batch string, yield func(rec BenchRecord, batch string) error) (BatchResult, error) {
	var res BatchResult
	cr := csv.NewReader(r)
	cr.FieldsPerRecord = -1 // a short or long row is handled per row, not as a fatal parse error
	cr.TrimLeadingSpace = true

	header, err := cr.Read()
	if err == io.EOF {
		return res, nil
	}
	if err != nil {
		return res, fmt.Errorf("benchimport: read csv header: %w", err)
	}
	colIndex := map[string]int{}
	for i, h := range header {
		colIndex[canonHeader(h)] = i
	}

	for {
		row, err := cr.Read()
		if err == io.EOF {
			break
		}
		if err != nil {
			// A structurally broken line is skipped, not fatal, so one bad row never drops the batch.
			res.Read++
			res.Skipped++
			continue
		}
		res.Read++
		get := func(key string) string {
			i, ok := colIndex[key]
			if !ok || i >= len(row) {
				return ""
			}
			return strings.TrimSpace(row[i])
		}
		rec := BenchRecord{
			FullName:      firstNonEmpty(get("name"), get("full_name"), get("fullname")),
			Location:      firstNonEmpty(get("location"), get("city_state")),
			JobTitle:      firstNonEmpty(get("job_title"), get("title"), get("role")),
			Skills:        splitList(firstNonEmpty(get("skills"), get("skill"))),
			Certs:         splitList(firstNonEmpty(get("certs"), get("certifications"), get("certification"))),
			Agencies:      splitList(firstNonEmpty(get("agencies"), get("agency"), get("customers"))),
			Years:         parseYears(firstNonEmpty(get("years"), get("years_experience"), get("experience"))),
			ClearanceCode: firstNonEmpty(get("clearance"), get("clearance_level")),
			ExternalID:    firstNonEmpty(get("external_id"), get("id")),
			Email:         strings.ToLower(firstNonEmpty(get("email"))),
			Phone:         firstNonEmpty(get("phone")),
		}
		if !hasSignal(rec) {
			res.Skipped++
			continue
		}
		// Count a record as mapped only after yield accepts it, so a batch cap (yield returns a stop error
		// before the record is processed) does not inflate the mapped count over new plus enriched.
		if err := yield(rec, batch); err != nil {
			return res, err
		}
		res.Mapped++
	}
	_ = ctx
	return res, nil
}

// jsonlRow is the decode shape for one JSONL object. The bench columns map to JSON keys with a few common
// aliases handled in fromJSONLRow.
type jsonlRow struct {
	Name           string   `json:"name"`
	FullName       string   `json:"full_name"`
	Location       string   `json:"location"`
	JobTitle       string   `json:"job_title"`
	Title          string   `json:"title"`
	Role           string   `json:"role"`
	Skills         []string `json:"skills"`
	Certs          []string `json:"certs"`
	Certifications []string `json:"certifications"`
	Agencies       []string `json:"agencies"`
	Years          int      `json:"years"`
	YearsExp       int      `json:"years_experience"`
	Clearance      string   `json:"clearance"`
	ClearanceLevel string   `json:"clearance_level"`
	ExternalID     string   `json:"external_id"`
	ID             string   `json:"id"`
	Email          string   `json:"email"`
	Phone          string   `json:"phone"`
}

// decodeJSONL reads one JSON object per line and yields one BenchRecord per well formed line. A blank line
// is ignored; a line that does not parse, or that yields no signal, is skipped and counted.
func decodeJSONL(ctx context.Context, r io.Reader, batch string, yield func(rec BenchRecord, batch string) error) (BatchResult, error) {
	var res BatchResult
	sc := bufio.NewScanner(r)
	sc.Buffer(make([]byte, 0, 64*1024), 4*1024*1024) // allow a fat profile line without splitting it
	for sc.Scan() {
		line := strings.TrimSpace(sc.Text())
		if line == "" {
			continue
		}
		res.Read++
		var jr jsonlRow
		if err := json.Unmarshal([]byte(line), &jr); err != nil {
			res.Skipped++
			continue
		}
		rec := fromJSONLRow(jr)
		if !hasSignal(rec) {
			res.Skipped++
			continue
		}
		// Count a record as mapped only after yield accepts it, so a batch cap (yield returns a stop error
		// before the record is processed) does not inflate the mapped count over new plus enriched.
		if err := yield(rec, batch); err != nil {
			return res, err
		}
		res.Mapped++
	}
	if err := sc.Err(); err != nil {
		return res, fmt.Errorf("benchimport: scan jsonl: %w", err)
	}
	_ = ctx
	return res, nil
}

// fromJSONLRow folds the alias keys into one BenchRecord.
func fromJSONLRow(jr jsonlRow) BenchRecord {
	return BenchRecord{
		FullName:      firstNonEmpty(strings.TrimSpace(jr.Name), strings.TrimSpace(jr.FullName)),
		Location:      strings.TrimSpace(jr.Location),
		JobTitle:      firstNonEmpty(strings.TrimSpace(jr.JobTitle), strings.TrimSpace(jr.Title), strings.TrimSpace(jr.Role)),
		Skills:        trimList(firstNonEmptyList(jr.Skills, nil)),
		Certs:         trimList(firstNonEmptyList(jr.Certs, jr.Certifications)),
		Agencies:      trimList(jr.Agencies),
		Years:         firstNonZero(jr.Years, jr.YearsExp),
		ClearanceCode: firstNonEmpty(strings.TrimSpace(jr.Clearance), strings.TrimSpace(jr.ClearanceLevel)),
		ExternalID:    firstNonEmpty(strings.TrimSpace(jr.ExternalID), strings.TrimSpace(jr.ID)),
		Email:         strings.ToLower(strings.TrimSpace(jr.Email)),
		Phone:         strings.TrimSpace(jr.Phone),
	}
}

// ---------- helpers ----------

// canonHeader normalizes a CSV header to a lowercased, underscore joined key so "Full Name", "full_name",
// and "FULL NAME" all map to the same column.
func canonHeader(h string) string {
	h = strings.ToLower(strings.TrimSpace(h))
	h = strings.ReplaceAll(h, " ", "_")
	return h
}

// splitList splits a delimited cell into trimmed tokens. The standard bench format uses a semicolon or a
// comma between skills, so both are accepted; an empty cell yields no tokens.
func splitList(s string) []string {
	if strings.TrimSpace(s) == "" {
		return nil
	}
	sep := ","
	if strings.Contains(s, ";") {
		sep = ";"
	}
	return trimList(strings.Split(s, sep))
}

// trimList trims each element and drops the empties.
func trimList(xs []string) []string {
	out := make([]string, 0, len(xs))
	for _, x := range xs {
		t := strings.TrimSpace(x)
		if t != "" {
			out = append(out, t)
		}
	}
	if len(out) == 0 {
		return nil
	}
	return out
}

func parseYears(s string) int {
	s = strings.TrimSpace(s)
	if s == "" {
		return 0
	}
	n, err := strconv.Atoi(s)
	if err != nil || n < 0 || n > 60 {
		return 0
	}
	return n
}

func firstNonEmpty(xs ...string) string {
	for _, x := range xs {
		if strings.TrimSpace(x) != "" {
			return strings.TrimSpace(x)
		}
	}
	return ""
}

func firstNonEmptyList(a, b []string) []string {
	if len(a) > 0 {
		return a
	}
	return b
}

func firstNonZero(xs ...int) int {
	for _, x := range xs {
		if x != 0 {
			return x
		}
	}
	return 0
}

// hasSignal reports whether a row carries enough structured signal to be worth writing. A row with no
// skills, no clearance, no certs, and no years is dropped so an empty or header only line is not imported.
func hasSignal(r BenchRecord) bool {
	return len(r.Skills) > 0 || r.ClearanceCode != "" || len(r.Certs) > 0 || r.Years > 0
}