package events

import (
	"context"
	"sync"
	"time"

	"github.com/alibaba/pouch/apis/types"
	"github.com/alibaba/pouch/pkg/log"

	goevents "github.com/docker/go-events"
	"github.com/pkg/errors"
)

const (
	eventsLimit = 64
)

// Events is pubsub channel for events generated by the engine.
type Events struct {
	mux sync.Mutex

	// support buffered events message
	events      []types.EventsMessage
	broadcaster *goevents.Broadcaster
}

// NewEvents return a new Events instance
func NewEvents() *Events {
	return &Events{
		events:      make([]types.EventsMessage, 0, eventsLimit),
		broadcaster: goevents.NewBroadcaster(),
	}
}

// Publish sends an event. The caller will be considered the initial
// publisher of the event. This means the timestamp will be calculated
// at this point and this method may read from the calling context.
func (e *Events) Publish(ctx context.Context, action string, eventType types.EventType, actor *types.EventsActor) error {
	// ensure actor not nil
	if actor == nil {
		actor = &types.EventsActor{}
	}

	now := time.Now().UTC()
	msg := types.EventsMessage{
		Action:   action,
		Type:     eventType,
		Actor:    actor,
		Time:     now.Unix(),
		TimeNano: now.UnixNano(),
	}

	// compatibility with moby
	switch eventType {
	case types.EventTypeContainer:
		msg.ID = actor.ID
		msg.Status = action
		if actor.Attributes != nil {
			if image, ok := actor.Attributes["image"]; ok {
				msg.From = image
			}
		}
	case types.EventTypeImage:
		msg.ID = actor.ID
		msg.Status = action
	}

	// put new event message to the buffer, if the numbers of messages
	// reach the buffer's limitation, discard the oldest event
	e.mux.Lock()
	if len(e.events) == cap(e.events) {
		// discard the oldest event
		copy(e.events, e.events[1:])
		e.events[len(e.events)-1] = msg
	} else {
		e.events = append(e.events, msg)
	}
	e.mux.Unlock()

	err := e.broadcaster.Write(&msg)
	if err != nil {
		log.With(ctx).Errorf("failed to publish event {action: %s, type: %s, id: %s}: %v", msg.Action, msg.Type, msg.ID, err)
	}

	return err
}

// Subscribe to events on the Events. Events are sent through the returned
// channel ch. If an error is encountered, it will be sent on channel errs and
// errs will be closed. To end the subscription, cancel the provided context.
//
// Zero or more filters may be provided as Args. Only events that match
// *any* of the provided filters will be sent on the channel.
func (e *Events) Subscribe(ctx context.Context, since, until time.Time, ef *Filter) ([]types.EventsMessage, <-chan *types.EventsMessage, <-chan error) {
	var (
		evch                  = make(chan *types.EventsMessage)
		errq                  = make(chan error, 1)
		channel               = goevents.NewChannel(0)
		queue                 = goevents.NewQueue(channel)
		dst     goevents.Sink = queue
	)

	closeAll := func() {
		close(errq)
		e.broadcaster.Remove(dst)
		queue.Close()
		channel.Close()
	}

	e.mux.Lock()
	buffered := e.filterBufferedEvents(since, until, ef)
	e.mux.Unlock()

	// add filters for event messages
	if ef != nil && ef.filter.Len() > 0 {
		dst = goevents.NewFilter(queue, goevents.MatcherFunc(func(gev goevents.Event) bool {
			// TODO(ziren): maybe we need adaptor here
			msg := gev.(*types.EventsMessage)
			return ef.Match(*msg)
		}))
	}

	e.broadcaster.Add(dst)

	go func() {
		defer closeAll()

		var err error
	loop:
		for {
			select {
			case ev := <-channel.C:
				env, ok := ev.(*types.EventsMessage)
				if !ok {
					err = errors.Errorf("invalid message encountered %#v; please file a bug", ev)
					break
				}

				select {
				case evch <- env:
				case <-ctx.Done():
					break loop
				}
			case <-ctx.Done():
				break loop
			}
		}

		if err == nil {
			if cerr := ctx.Err(); cerr != context.Canceled {
				err = cerr
			}
		}

		errq <- err
	}()

	return buffered, evch, errq
}

// filterBufferedEvents iterates over the cached events in the buffer
// and returns those that were emitted between two specific dates.
func (e *Events) filterBufferedEvents(since, until time.Time, ef *Filter) []types.EventsMessage {
	var buffered []types.EventsMessage
	if since.IsZero() && until.IsZero() {
		return buffered
	}

	var sinceNanoUnix int64
	if !since.IsZero() {
		sinceNanoUnix = since.UnixNano()
	}

	var untilNanoUnix int64
	if !until.IsZero() {
		untilNanoUnix = until.UnixNano()
	}

	for i := len(e.events) - 1; i >= 0; i-- {
		ev := e.events[i]

		if ev.TimeNano < sinceNanoUnix {
			break
		}

		if untilNanoUnix > 0 && ev.TimeNano > untilNanoUnix {
			continue
		}

		if ef == nil || ef.Match(ev) {
			// keep the order of messages
			buffered = append([]types.EventsMessage{ev}, buffered...)
		}
	}
	return buffered
}
