orca/internal/jobspec/markdown.go

package jobspec

import (
	"fmt"
	"strconv"
	"strings"
)

// WorkloadSpec is the unified canonical jobspec populated by both the
// Markdown frontmatter parser (canonical path, R-013/R-014) and the HCL
// legacy adapter (REQ-064, REQ-090). It is the single shape consumed by
// downstream phases (P0c schemas, P01 transport). The Markdown body
// after the closing `---` is preserved verbatim in Body (R-015
// byte-exact preservation is a load-bearing invariant enforced by the
// fuzz harness in markdown_fuzz_test.go).
type WorkloadSpec struct {
	SpecVersion string
	Kind        string
	Name        string
	Runtime     *RuntimeBlock
	Count       int
	Ports       []PortSpec
	Env         map[string]string
	Secrets     []string
	Volumes     []VolumeSpec
	Body        string

	// Kind-specific blocks consumed by the P0c schema validators
	// (internal/spec/schema). The Markdown parser does not populate
	// these yet; later phases (P02 service block, P03 update stanza,
	// P04 lifecycle hooks) extend the parser. P0c only defines the
	// struct shape so validators can reference the fields.

	// Restart is the restart policy block. Required for Service and
	// DaemonSet; optional for Job (defaults to never/on-failure).
	// Populated by the P04 lifecycle phase.
	Restart *RestartBlock

	// Schedule is the schedule block. For Job it carries an optional
	// cron string; for DaemonSet it carries the placement mode
	// (every-node/matching/mandatory). Populated by P05 (scheduler
	// skeleton) and the DaemonSet phase.
	Schedule *ScheduleBlock

	// Update is the rolling/canary update stanza. Required for
	// Service. Populated by the P03 update-stanza phase.
	Update *UpdateBlock

	// Service is the service block (Traefik route definition). For
	// Service kind it is implied; Job and DaemonSet do not carry a
	// Traefik route by default (D-175). Populated by the P02 service
	// block phase.
	Service *ServiceBlock

	// Timeout is an optional execution timeout (duration string) for
	// Job. Populated by P04.
	Timeout string
}

// RuntimeBlock is a minimal runtime abstraction surface populated by the
// Markdown parser. The full runtime abstraction lands in P07; for now
// only the one_of/image/command fields are parsed and stored (REQ-064).
type RuntimeBlock struct {
	OneOf   string
	Image   string
	Command string
}

// RestartBlock is the restart policy block. Mode is one of never,
// on-failure, service (REQ-074 schema validators). Populated by P04.
type RestartBlock struct {
	Mode       string
	MaxRetries int
	Delay      string
}

// ScheduleBlock is the scheduling block. For Job, Cron is an optional
// cron expression. For DaemonSet, Mode is one of every-node, matching,
// mandatory (REQ-074). Populated by P05 and the DaemonSet phase.
type ScheduleBlock struct {
	Mode string
	Cron string
}

// UpdateBlock is the rolling/canary update stanza. Required for Service.
// Populated by P03.
type UpdateBlock struct {
	Strategy string
	MaxSurge int
}

// ServiceBlock is the Traefik route definition. For Service it is
// implied (Traefik route YES); Job and DaemonSet do not carry one by
// default (D-175). Populated by P02.
type ServiceBlock struct {
	Host    string
	RouteID string
}

// PortSpec is a minimal port binding entry. HostIP is optional.
type PortSpec struct {
	Name     string
	HostPort int
	Port     int
	Protocol string
	HostIP   string
}

// VolumeSpec is a minimal volume mount entry. Fields are stored raw
// pending the P0c schema work (REQ-074).
type VolumeSpec struct {
	Name     string
	Type     string
	Source   string
	Target   string
	ReadOnly bool
}

// validKinds is the set of workload kinds accepted by the parser per
// R-012. Unknown kinds are rejected.
var validKinds = map[string]bool{
	"Job":       true,
	"Service":   true,
	"DaemonSet": true,
}

// ParseMarkdown parses a Markdown jobspec with YAML frontmatter into a
// *WorkloadSpec (R-013 canonical format, R-014 frontmatter). The body
// after the closing `---` is preserved verbatim in result.Body
// (R-015 byte-exact, including trailing newlines, CRLF, and BOM in the
// body). The frontmatter parser is a minimal hand-rolled YAML-ish
// key:value reader — gopkg.in/yaml.v3 is intentionally not added (same
// approach as internal/config/markdown.go and internal/ns/parse.go).
//
// For .yaml/.yml files (no Markdown body), the dispatcher calls this
// with the whole file treated as frontmatter and Body left empty (see
// dispatch.go).
func ParseMarkdown(data []byte) (*WorkloadSpec, error) {
	content := string(data)
	block, body, ok := splitFrontmatter(content)
	if !ok {
		return nil, fmt.Errorf("parse markdown: missing frontmatter delimiters")
	}
	if strings.TrimSpace(block) == "" {
		return nil, fmt.Errorf("parse markdown: empty frontmatter")
	}
	spec, err := parseFrontmatterBlock(block)
	if err != nil {
		return nil, err
	}
	spec.Body = body
	if err := validateWorkload(spec); err != nil {
		return nil, err
	}
	return spec, nil
}

// splitFrontmatter splits the file content into the YAML frontmatter
// block and the verbatim body that follows the closing `---`. A leading
// UTF-8 BOM is stripped from the frontmatter scan (R-015: BOM is not
// preserved in the frontmatter, but a BOM inside the body would be
// preserved because the body is verbatim). Returns (block, body, ok).
// ok is false when no opening `---` delimiter is present, or no closing
// `---` delimiter is found, or the block is empty after the opening
// delimiter (handled by caller).
func splitFrontmatter(content string) (block, body string, ok bool) {
	// Strip a leading UTF-8 BOM if present (EF BB BF). Only the
	// frontmatter scan is BOM-stripped; the body is byte-exact, so a BOM
	// appearing inside the body is preserved verbatim.
	stripped := content
	if strings.HasPrefix(stripped, "\uFEFF") {
		stripped = stripped[len("\uFEFF"):]
	}
	// Trim leading horizontal whitespace and newlines before the
	// opening delimiter. We do NOT trim trailing — body must be exact.
	trimmed := strings.TrimLeft(stripped, "\r\n\t ")
	if !strings.HasPrefix(trimmed, "---") {
		return "", "", false
	}
	// The opening delimiter must be on its own line: `---` optionally
	// followed by a line terminator.
	rest := trimmed[3:]
	// The opening `---` must be followed by a newline or end-of-file
	// (a `---foo` prefix is not a valid delimiter).
	if len(rest) > 0 && rest[0] != '\n' && rest[0] != '\r' {
		return "", "", false
	}
	rest = strings.TrimLeft(rest, "\r\n")

	// Find the closing delimiter line. The closing `---` must be on its
	// own line: preceded by a newline (or at the start of `rest`) and
	// followed by a newline or end-of-file.
	idx := findClosingDelimiter(rest)
	if idx < 0 {
		return "", "", false
	}
	block = rest[:idx]
	// Body is everything after the closing `---` line. The closing
	// delimiter line itself (including its trailing newline) is NOT
	// part of the body. We compute the byte offset in the original
	// `content` so the body is byte-exact (R-015).
	afterClose := rest[idx:]
	// afterClose starts with `---`. Strip the delimiter line.
	delimLen := 3
	// Account for an optional trailing `...` or spaces on the delimiter
	// line — the delimiter is `---` followed by anything up to and
	// including the line terminator. Body starts after the newline.
	// Find the end of the delimiter line.
	newlineIdx := strings.IndexAny(afterClose, "\r\n")
	var bodyStart int
	if newlineIdx < 0 {
		// Closing `---` is the last line: body is empty.
		bodyStart = len(afterClose)
	} else {
		// Consume the delimiter line and its line terminator(s).
		bodyStart = newlineIdx
		// Strip a single CRLF or LF.
		if strings.HasPrefix(afterClose[bodyStart:], "\r\n") {
			bodyStart += 2
		} else {
			bodyStart += 1
		}
	}
	body = afterClose[bodyStart:]
	_ = delimLen
	return block, body, true
}

// findClosingDelimiter returns the byte index in `rest` where the
// closing `---` delimiter line begins, or -1 if none is found. The
// delimiter must be on its own line: either at the start of `rest` or
// preceded by a newline, and followed by a newline or end-of-file.
func findClosingDelimiter(rest string) int {
	// Special case: closing delimiter at the very start (frontmatter
	// block is empty). The opening `---` is immediately followed by the
	// closing `---`. We require the opening to be its own line, so the
	// closing at index 0 means the opening had no body — invalid (empty
	// frontmatter handled by caller). We still report it; caller
	// rejects empty block.
	for i := 0; i < len(rest); i++ {
		if rest[i] != '\n' {
			continue
		}
		// Candidate: line after this newline starts with `---`.
		j := i + 1
		if j+3 <= len(rest) && rest[j] == '-' && rest[j+1] == '-' && rest[j+2] == '-' {
			// Must be followed by newline, CRLF, or end-of-file.
			end := j + 3
			if end == len(rest) {
				return j
			}
			if rest[end] == '\n' || rest[end] == '\r' {
				return j
			}
		}
	}
	// Final candidate: closing delimiter at the very start of rest
	// (immediately after the opening delimiter + its newline). This
	// happens when frontmatter is empty: `---\n---\n`. We already trim
	// leading newlines off `rest`, so if rest itself starts with `---`
	// AND it's a closing delimiter (followed by newline/EOF), it is the
	// empty-frontmatter case.
	if strings.HasPrefix(rest, "---") {
		end := 3
		if end == len(rest) {
			return 0
		}
		if rest[end] == '\n' || rest[end] == '\r' {
			return 0
		}
	}
	return -1
}

// parseFrontmatterBlock parses a minimal YAML-ish frontmatter block into
// a *WorkloadSpec (without Body, which is filled by the caller).
//
// Supported shapes:
//
//	kind: Job
//	name: my-job
//	count: 3
//	runtime:
//	  one_of: process
//	  image: docker.io/nginx:latest
//	  command: /bin/sh -c
//	ports:
//	  - name: http
//	    port: 8080
//	    host_port: 80
//	    protocol: tcp
//	env:
//	  FOO: bar
//	  BAR:
//	    from: secret:my-secret
//	secrets:
//	  - db-password
//	volumes:
//	  - name: data
//	    type: host
//	    source: /data
//	    target: /data
//	    read_only: true
//
// Comments (# ...) and blank lines are ignored. Quoted scalar values
// ("..." or '...') are unwrapped. No flow collections except the
// inline-array form for `secrets`. Multi-line block scalars (|, >) are
// not supported — by design, to avoid adding a YAML dependency for this
// small surface.
func parseFrontmatterBlock(block string) (*WorkloadSpec, error) {
	spec := &WorkloadSpec{Count: 1}
	lines := strings.Split(block, "\n")

	type section int
	const (
		secNone section = iota
		secRuntime
		secPorts
		secEnv
		secSecrets
		secVolumes
	)
	cur := secNone
	var curPort *PortSpec
	var curVol *VolumeSpec

	flushPort := func() {
		if curPort != nil {
			spec.Ports = append(spec.Ports, *curPort)
			curPort = nil
		}
	}
	flushVol := func() {
		if curVol != nil {
			spec.Volumes = append(spec.Volumes, *curVol)
			curVol = nil
		}
	}

	for lineNo, raw := range lines {
		line := stripComment(raw)
		if strings.TrimSpace(line) == "" {
			continue
		}
		indent := countIndent(line)
		trimmed := strings.TrimSpace(line)

		if indent == 0 {
			// Flush any pending nested entry before switching sections.
			flushPort()
			flushVol()
			cur = secNone

			key, val, ok := splitKV(trimmed)
			if !ok {
				return nil, fmt.Errorf("parse markdown: line %d: malformed key:value", lineNo+1)
			}
			switch key {
			case "orca-spec-version":
				spec.SpecVersion = unquote(val)
			case "kind":
				spec.Kind = unquote(val)
			case "name":
				spec.Name = unquote(val)
			case "count":
				if n, err := strconv.Atoi(strings.TrimSpace(unquote(val))); err == nil {
					spec.Count = n
				} else {
					return nil, fmt.Errorf("parse markdown: line %d: count: %v", lineNo+1, err)
				}
			case "runtime":
				spec.Runtime = &RuntimeBlock{}
				if strings.TrimSpace(val) != "" {
					// Inline value (unusual); ignore — runtime is a block.
				}
				cur = secRuntime
			case "ports":
				cur = secPorts
			case "env":
				spec.Env = map[string]string{}
				cur = secEnv
			case "secrets":
				if strings.TrimSpace(val) != "" {
					arr, err := parseStringArray(val)
					if err != nil {
						return nil, fmt.Errorf("parse markdown: line %d: secrets: %w", lineNo+1, err)
					}
					spec.Secrets = append(spec.Secrets, arr...)
					cur = secNone
				} else {
					cur = secSecrets
				}
			case "volumes":
				cur = secVolumes
			default:
				// Unknown top-level keys are ignored (forward-compat).
				cur = secNone
			}
			continue
		}

		// Indented line: a nested entry under the current section.
		switch cur {
		case secRuntime:
			if spec.Runtime == nil {
				spec.Runtime = &RuntimeBlock{}
			}
			key, val, ok := splitKV(trimmed)
			if !ok {
				continue
			}
			switch key {
			case "one_of":
				spec.Runtime.OneOf = unquote(val)
			case "image":
				spec.Runtime.Image = unquote(val)
			case "command":
				spec.Runtime.Command = unquote(val)
			}
		case secPorts:
			if strings.HasPrefix(trimmed, "- ") || trimmed == "-" {
				flushPort()
				p := PortSpec{}
				curPort = &p
				rest := strings.TrimSpace(strings.TrimPrefix(trimmed, "-"))
				if rest != "" {
					applyPortKV(curPort, rest)
				}
			} else if curPort != nil {
				applyPortKV(curPort, trimmed)
			}
		case secEnv:
			key, val, ok := splitKV(trimmed)
			if !ok {
				continue
			}
			if val == "" {
				// Nested mapping under env (e.g. `BAR:\n  from: ...`).
				// Store the raw string for now (REQ-064: store raw).
				spec.Env[key] = ""
			} else if strings.HasPrefix(val, "{") && strings.HasSuffix(val, "}") {
				// Inline object form: `BAR: {from: "secret:..."}`.
				// Store the raw object string for now.
				spec.Env[key] = val
			} else {
				spec.Env[key] = unquote(val)
			}
		case secSecrets:
			if strings.HasPrefix(trimmed, "- ") || trimmed == "-" {
				item := strings.TrimSpace(strings.TrimPrefix(trimmed, "-"))
				if item != "" {
					spec.Secrets = append(spec.Secrets, unquote(item))
				}
			}
		case secVolumes:
			if strings.HasPrefix(trimmed, "- ") || trimmed == "-" {
				flushVol()
				v := VolumeSpec{}
				curVol = &v
				rest := strings.TrimSpace(strings.TrimPrefix(trimmed, "-"))
				if rest != "" {
					applyVolumeKV(curVol, rest)
				}
			} else if curVol != nil {
				applyVolumeKV(curVol, trimmed)
			}
		}
	}
	flushPort()
	flushVol()
	return spec, nil
}

// applyPortKV applies a `key: value` pair to a PortSpec entry.
func applyPortKV(p *PortSpec, s string) {
	key, val, ok := splitKV(s)
	if !ok {
		return
	}
	switch key {
	case "name":
		p.Name = unquote(val)
	case "host_port":
		if n, err := strconv.Atoi(strings.TrimSpace(unquote(val))); err == nil {
			p.HostPort = n
		}
	case "port":
		if n, err := strconv.Atoi(strings.TrimSpace(unquote(val))); err == nil {
			p.Port = n
		}
	case "protocol":
		p.Protocol = unquote(val)
	case "host_ip":
		p.HostIP = unquote(val)
	}
}

// applyVolumeKV applies a `key: value` pair to a VolumeSpec entry.
func applyVolumeKV(v *VolumeSpec, s string) {
	key, val, ok := splitKV(s)
	if !ok {
		return
	}
	switch key {
	case "name":
		v.Name = unquote(val)
	case "type":
		v.Type = unquote(val)
	case "source":
		v.Source = unquote(val)
	case "target":
		v.Target = unquote(val)
	case "read_only":
		switch strings.ToLower(strings.TrimSpace(unquote(val))) {
		case "true", "yes", "on", "1":
			v.ReadOnly = true
		}
	}
}

// validateWorkload enforces required fields and kind validity (R-012).
func validateWorkload(spec *WorkloadSpec) error {
	if spec.Kind == "" {
		return fmt.Errorf("parse markdown: missing kind")
	}
	if !validKinds[spec.Kind] {
		return fmt.Errorf("parse markdown: kind %q is not one of Job, Service, DaemonSet", spec.Kind)
	}
	if strings.TrimSpace(spec.Name) == "" {
		return fmt.Errorf("parse markdown: missing name")
	}
	return nil
}

// parseStringArray parses an inline YAML flow-array of scalars, e.g.
// `["a", "b"]` or `['a', 'b']` or `[a, b]`. Empty array `[]` returns nil.
func parseStringArray(val string) ([]string, error) {
	val = strings.TrimSpace(val)
	if val == "" {
		return nil, nil
	}
	if !strings.HasPrefix(val, "[") || !strings.HasSuffix(val, "]") {
		return nil, fmt.Errorf("expected [..] array, got %q", val)
	}
	inner := strings.TrimSpace(val[1 : len(val)-1])
	if inner == "" {
		return nil, nil
	}
	parts := splitFlowItems(inner)
	out := make([]string, 0, len(parts))
	for _, p := range parts {
		p = strings.TrimSpace(p)
		if p == "" {
			continue
		}
		out = append(out, unquote(p))
	}
	return out, nil
}

// splitFlowItems splits a comma-separated flow-array body, respecting
// single and double quotes.
func splitFlowItems(s string) []string {
	var out []string
	inSingle := false
	inDouble := false
	start := 0
	for i := 0; i < len(s); i++ {
		c := s[i]
		switch c {
		case '\'':
			if !inDouble {
				inSingle = !inSingle
			}
		case '"':
			if !inSingle {
				inDouble = !inDouble
			}
		case ',':
			if !inSingle && !inDouble {
				out = append(out, s[start:i])
				start = i + 1
			}
		}
	}
	out = append(out, s[start:])
	return out
}

func countIndent(s string) int {
	n := 0
	for _, r := range s {
		if r == ' ' || r == '\t' {
			n++
			continue
		}
		break
	}
	return n
}

func splitKV(s string) (key, val string, ok bool) {
	idx := strings.Index(s, ":")
	if idx < 0 {
		return "", "", false
	}
	key = strings.TrimSpace(s[:idx])
	val = strings.TrimSpace(s[idx+1:])
	if key == "" {
		return "", "", false
	}
	return key, val, true
}

func stripComment(s string) string {
	inSingle := false
	inDouble := false
	for i := 0; i < len(s); i++ {
		c := s[i]
		switch c {
		case '\'':
			if !inDouble {
				inSingle = !inSingle
			}
		case '"':
			if !inSingle {
				inDouble = !inDouble
			}
		case '#':
			if !inSingle && !inDouble {
				if i == 0 || s[i-1] == ' ' || s[i-1] == '\t' {
					return s[:i]
				}
			}
		}
	}
	return s
}

func unquote(s string) string {
	s = strings.TrimSpace(s)
	if len(s) >= 2 {
		if (s[0] == '"' && s[len(s)-1] == '"') || (s[0] == '\'' && s[len(s)-1] == '\'') {
			return s[1 : len(s)-1]
		}
	}
	return s
}