//go:build windows

package gcs

import (
	"context"
	"errors"
	"fmt"
	"sync"
	"time"

	"github.com/Microsoft/hcsshim/internal/cow"
	"github.com/Microsoft/hcsshim/internal/gcs/prot"
	"github.com/Microsoft/hcsshim/internal/hcs/schema1"
	hcsschema "github.com/Microsoft/hcsshim/internal/hcs/schema2"
	"github.com/Microsoft/hcsshim/internal/jobobject"
	"github.com/Microsoft/hcsshim/internal/log"
	"github.com/Microsoft/hcsshim/internal/ot"
	"github.com/containerd/errdefs"
	"go.opentelemetry.io/otel/attribute"
)

const hrComputeSystemDoesNotExist = 0xc037010e

// Container implements the cow.Container interface for containers
// created via GuestConnection.
type Container struct {
	gc        *GuestConnection
	id        string
	notifyCh  chan struct{}
	closeCh   chan struct{}
	closeOnce sync.Once
	// waitBlock is the channel used to wait for container shutdown or termination
	waitBlock chan struct{}
	// waitError indicates the container termination error if any
	waitError error
}

var _ cow.Container = &Container{}

// CreateContainer creates a container using ID `cid` and `cfg`. The request
// will likely not be cancellable even if `ctx` becomes done.
func (gc *GuestConnection) CreateContainer(ctx context.Context, cid string, config interface{}) (_ *Container, err error) {
	ctx, span := ot.StartSpan(ctx, "gcs::GuestConnection::CreateContainer", ot.WithClientSpanKind)
	defer span.End()
	defer func() { ot.SetSpanStatus(span, err) }()
	span.SetAttributes(attribute.String("cid", cid))

	c := &Container{
		gc:        gc,
		id:        cid,
		notifyCh:  make(chan struct{}),
		closeCh:   make(chan struct{}),
		waitBlock: make(chan struct{}),
	}
	err = gc.requestNotify(cid, c.notifyCh)
	if err != nil {
		return nil, err
	}
	req := prot.ContainerCreate{
		RequestBase:     makeRequest(ctx, cid),
		ContainerConfig: prot.AnyInString{Value: config},
	}
	var resp prot.ContainerCreateResponse
	err = gc.brdg.RPC(ctx, prot.RPCCreate, &req, &resp, false)
	if err != nil {
		return nil, err
	}
	go c.waitBackground()
	return c, nil
}

// OpenContainer attaches a host-side wrapper to a container already
// running inside the UVM.
func (gc *GuestConnection) OpenContainer(_ context.Context, cid string) (_ *Container, err error) {
	c := &Container{
		gc:        gc,
		id:        cid,
		notifyCh:  make(chan struct{}),
		closeCh:   make(chan struct{}),
		waitBlock: make(chan struct{}),
	}
	err = gc.requestNotify(cid, c.notifyCh)
	if err != nil {
		return nil, err
	}
	go c.waitBackground()
	return c, nil
}

// OS returns the operating system of the container, "linux" or "windows".
func (c *Container) OS() string {
	return c.gc.os
}

// IsOCI specifies whether CreateProcess should be called with an OCI
// specification in its input.
func (c *Container) IsOCI() bool {
	return c.gc.os != "windows"
}

// Close releases associated with the container.
func (c *Container) Close() error {
	c.closeOnce.Do(func() {
		_, span := ot.StartSpan(context.Background(), "gcs::Container::Close")
		defer span.End()
		span.SetAttributes(attribute.String("cid", c.id))

		close(c.closeCh)
	})
	return nil
}

// CreateProcess creates a process in the container.
func (c *Container) CreateProcess(ctx context.Context, config interface{}) (_ cow.Process, err error) {
	ctx, span := ot.StartSpan(ctx, "gcs::Container::CreateProcess", ot.WithClientSpanKind)
	defer span.End()
	defer func() { ot.SetSpanStatus(span, err) }()
	span.SetAttributes(attribute.String("cid", c.id))

	return c.gc.exec(ctx, c.id, config)
}

// OpenProcessWithIO is the live-migration restore counterpart of
// [Container.CreateProcess]: it attaches to a process already running
// in this container, re-listens on the supplied vsock ports, and
// pre-registers the source bridge's WaitForProcess id so the guest's
// still-outstanding response is routed.
// watchExit launches the background exit wait, first probing for an already-
// exited process whose adopted response the guest never redelivers; pass false
// when the caller already has one outstanding (a source rollback reuses the
// live process's).
func (c *Container) OpenProcessWithIO(ctx context.Context, pid uint32, stdinPort, stdoutPort, stderrPort uint32, waitCallID int64, watchExit bool) (_ *Process, err error) {
	ctx, span := ot.StartSpan(ctx, "gcs::Container::OpenProcessWithIO", ot.WithClientSpanKind)
	defer span.End()
	defer func() { ot.SetSpanStatus(span, err) }()
	span.SetAttributes(
		attribute.String("cid", c.id),
		attribute.Int64("pid", int64(pid)),
		attribute.Int64("waitCallID", waitCallID))

	if waitCallID == 0 {
		return nil, fmt.Errorf("open process pid %d in container %s: waitCallID is required", pid, c.id)
	}

	p := &Process{
		gc:         c.gc,
		cid:        c.id,
		id:         pid,
		stdinPort:  stdinPort,
		stdoutPort: stdoutPort,
		stderrPort: stderrPort,
	}
	defer func() {
		if err != nil {
			p.Close()
		}
	}()

	listen := func(port uint32) (*ioChannel, error) {
		if port == 0 {
			return nil, nil
		}
		l, err := c.gc.ioListenFn(port)
		if err != nil {
			return nil, fmt.Errorf("listen vsock port %d: %w", port, err)
		}
		return newIoChannel(l), nil
	}
	if p.stdin, err = listen(stdinPort); err != nil {
		return nil, err
	}
	if p.stdout, err = listen(stdoutPort); err != nil {
		return nil, err
	}
	if p.stderr, err = listen(stderrPort); err != nil {
		return nil, err
	}

	p.waitResp = &prot.ContainerWaitForProcessResponse{}
	p.waitCall, err = c.gc.brdg.PreregisterRPC(waitCallID, prot.RPCWaitForProcess, p.waitResp)
	if err != nil {
		return nil, fmt.Errorf("preregister wait for pid %d in container %s (id %d): %w", pid, c.id, waitCallID, err)
	}
	// A reused outstanding wait carries its own response; adopt it so the
	// reported exit code reflects the guest's reply and not this unused one.
	if resp, ok := p.waitCall.resp.(*prot.ContainerWaitForProcessResponse); ok {
		p.waitResp = resp
	}
	// Establish exit reporting unless the caller already watches this process
	// (a reuse that already has a wait outstanding, e.g. a rollback).
	if watchExit {
		p.startExitWatch(ctx)
	}
	log.G(ctx).WithField("pid", p.id).Debug("opened existing process with IO")
	return p, nil
}

// ID returns the container's ID.
func (c *Container) ID() string {
	return c.id
}

// Modify sends a modify request to the container.
func (c *Container) Modify(ctx context.Context, config interface{}) (err error) {
	ctx, span := ot.StartSpan(ctx, "gcs::Container::Modify", ot.WithClientSpanKind)
	defer span.End()
	defer func() { ot.SetSpanStatus(span, err) }()
	span.SetAttributes(attribute.String("cid", c.id))

	req := prot.ContainerModifySettings{
		RequestBase: makeRequest(ctx, c.id),
		Request:     config,
	}
	var resp prot.ResponseBase
	return c.gc.brdg.RPC(ctx, prot.RPCModifySettings, &req, &resp, false)
}

// SetCPUGroupAffinities implements the cow.Container interface. CPU affinity is
// applied on the host's silo job object, which a guest-side container does not
// own, so this returns ErrNotImplemented.
func (c *Container) SetCPUGroupAffinities(_ context.Context, _ []jobobject.GroupAffinity) error {
	return fmt.Errorf("cpu affinity is not supported for guest containers: %w", errdefs.ErrNotImplemented)
}

// Properties returns the requested container properties targeting a V1 schema prot.Container.
func (c *Container) Properties(ctx context.Context, types ...schema1.PropertyType) (_ *schema1.ContainerProperties, err error) {
	ctx, span := ot.StartSpan(ctx, "gcs::Container::Properties", ot.WithClientSpanKind)
	defer span.End()
	defer func() { ot.SetSpanStatus(span, err) }()
	span.SetAttributes(attribute.String("cid", c.id))

	req := prot.ContainerGetProperties{
		RequestBase: makeRequest(ctx, c.id),
		Query:       prot.ContainerPropertiesQuery{PropertyTypes: types},
	}
	var resp prot.ContainerGetPropertiesResponse
	err = c.gc.brdg.RPC(ctx, prot.RPCGetProperties, &req, &resp, true)
	if err != nil {
		return nil, err
	}
	return (*schema1.ContainerProperties)(&resp.Properties), nil
}

// PropertiesV2 returns the requested container properties targeting a V2 schema container.
func (c *Container) PropertiesV2(ctx context.Context, types ...hcsschema.PropertyType) (_ *hcsschema.Properties, err error) {
	ctx, span := ot.StartSpan(ctx, "gcs::Container::PropertiesV2", ot.WithClientSpanKind)
	defer span.End()
	defer func() { ot.SetSpanStatus(span, err) }()
	span.SetAttributes(attribute.String("cid", c.id))

	req := prot.ContainerGetPropertiesV2{
		RequestBase: makeRequest(ctx, c.id),
		Query:       prot.ContainerPropertiesQueryV2{PropertyTypes: types},
	}
	var resp prot.ContainerGetPropertiesResponseV2
	err = c.gc.brdg.RPC(ctx, prot.RPCGetProperties, &req, &resp, true)
	if err != nil {
		return nil, err
	}
	return (*hcsschema.Properties)(&resp.Properties), nil
}

// Start starts the container.
func (c *Container) Start(ctx context.Context) (err error) {
	ctx, span := ot.StartSpan(ctx, "gcs::Container::Start", ot.WithClientSpanKind)
	defer span.End()
	defer func() { ot.SetSpanStatus(span, err) }()
	span.SetAttributes(attribute.String("cid", c.id))

	req := makeRequest(ctx, c.id)
	var resp prot.ResponseBase
	return c.gc.brdg.RPC(ctx, prot.RPCStart, &req, &resp, false)
}

func (c *Container) shutdown(ctx context.Context, proc prot.RPCProc) error {
	req := makeRequest(ctx, c.id)
	var resp prot.ResponseBase
	err := c.gc.brdg.RPC(ctx, proc, &req, &resp, true)
	if err != nil {
		if uint32(resp.Result) != hrComputeSystemDoesNotExist {
			return err
		}
		select {
		case <-c.notifyCh:
		default:
			log.G(ctx).WithError(err).Warn("ignoring missing container")
		}
	}
	return nil
}

// Shutdown sends a graceful shutdown request to the container. The container
// might not be terminated by the time the request completes (and might never
// terminate).
func (c *Container) Shutdown(ctx context.Context) (err error) {
	ctx, span := ot.StartSpan(ctx, "gcs::Container::Shutdown", ot.WithClientSpanKind)
	defer span.End()
	defer func() { ot.SetSpanStatus(span, err) }()
	span.SetAttributes(attribute.String("cid", c.id))

	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	return c.shutdown(ctx, prot.RPCShutdownGraceful)
}

// Terminate sends a forceful terminate request to the container. The container
// might not be terminated by the time the request completes (and might never
// terminate).
func (c *Container) Terminate(ctx context.Context) (err error) {
	ctx, span := ot.StartSpan(ctx, "gcs::Container::Terminate", ot.WithClientSpanKind)
	defer span.End()
	defer func() { ot.SetSpanStatus(span, err) }()
	span.SetAttributes(attribute.String("cid", c.id))

	ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
	defer cancel()
	return c.shutdown(ctx, prot.RPCShutdownForced)
}

func (c *Container) WaitChannel() <-chan struct{} {
	return c.waitBlock
}

func (c *Container) WaitError() error {
	return c.waitError
}

// Wait waits for the container to terminate (or Close to be called, or the
// guest connection to terminate).
func (c *Container) Wait() error {
	<-c.WaitChannel()
	return c.WaitError()
}

func (c *Container) waitBackground() {
	ctx, span := ot.StartSpan(context.Background(), "gcs::Container::waitBackground")
	defer span.End()
	span.SetAttributes(attribute.String("cid", c.id))

	select {
	case <-c.notifyCh:
	case <-c.closeCh:
		c.waitError = errors.New("container closed")
	}
	close(c.waitBlock)

	log.G(ctx).Debug("container exited")
	ot.SetSpanStatus(span, c.waitError)
}