package autoradio

import (
	"bytes"
	"crypto/rand"
	"encoding/base64"
	"encoding/json"
	"errors"
	"strings"
	"sync"
	"time"

	"git.autistici.org/ale/autoradio/third_party/github.com/coreos/go-etcd/etcd"
)

var (
	MasterElectionPath = "/icecast/cluster/master"
	MountPrefix        = "/icecast/mounts/"
	NodePrefix         = "/icecast/nodes/"

	IcecastPort = 8000

	ErrIsDirectory = errors.New("key is a directory")
	ErrIsFile      = errors.New("key is a file")
)

// A mountpoint for a stream.
type Mount struct {
	// Name (path to the mountpoint).
	Name string

	// Username for source authentication.
	Username string

	// Password for source authentication.
	Password string

	// Is this field is set, the mount point will be relaying an
	// external stream and no source connections will be accepted.
	// Each node will pull from the external source directly,
	// ignoring our master election protocol.
	RelayUrl string

	// Fallback stream name (optional).
	Fallback string
}

func (m *Mount) IsRelay() bool {
	return m.RelayUrl != ""
}

func mountPath(mountName string) string {
	return MountPrefix + mountName[1:]
}

// Status of a mount on an individual Icecast server.
type IcecastMountStatus struct {
	Name         string
	Listeners    int
	BitRate      int
	Quality      float64
	VideoQuality float64
	FrameSize    string
	FrameRate    float64
}

// Status of a node. This is used to report load and stream status.
type NodeStatus struct {
	// Public IP of this server.
	IP string

	// Is the Icecast server up?
	IcecastUp bool

	// List of mount points.
	Mounts []IcecastMountStatus `xml:"mount"`

	// Bandwidth utilization.
	BandwidthUsage float64
}

func (ns *NodeStatus) NumListeners() int {
	listeners := 0
	for _, m := range ns.Mounts {
		listeners += m.Listeners
	}
	return listeners
}

// Cache the list of active nodes (the front-ends that need to
// retrieve this information continuously, so we limit them to 2qps).
type nodesCache struct {
	ttl      time.Duration
	nodes    []*NodeStatus
	deadline time.Time
	lock     sync.Mutex
}

type getNodesFunc func() ([]*NodeStatus, error)

func newNodesCache() *nodesCache {
	return &nodesCache{
		ttl: 500 * time.Millisecond,
	}
}

// Get returns the cached value of 'fn', if valid. If the value is
// expired and we get an error from 'fn', we will attempt to return
// the previously cached value anyway, along with the error: the
// caller can then pick the right failure behavior.
func (nc *nodesCache) Get(fn getNodesFunc) ([]*NodeStatus, error) {
	nc.lock.Lock()
	defer nc.lock.Unlock()

	var err error
	now := time.Now()
	if now.After(nc.deadline) {
		var nodes []*NodeStatus
		if nodes, err = fn(); err == nil {
			nc.nodes = nodes
			nc.deadline = now.Add(nc.ttl)
		}
	}
	return nc.nodes, err
}

// RadioAPI is the actual API to the streaming cluster's database.
type RadioAPI struct {
	client           *etcd.Client
	activeNodesCache *nodesCache
}

func NewRadioAPI(client *etcd.Client) *RadioAPI {
	return &RadioAPI{client, newNodesCache()}
}

// GetMount returns data on a specific mountpoint (returns nil if not
// found).
func (r *RadioAPI) GetMount(mountName string) (*Mount, error) {
	response, err := r.client.Get(mountPath(mountName), false, false)
	if err != nil || response.Node == nil {
		return nil, err
	}
	if response.Node.Dir {
		return nil, ErrIsDirectory
	}

	var m Mount
	if err := json.NewDecoder(strings.NewReader(response.Node.Value)).Decode(&m); err != nil {
		return nil, err
	}
	return &m, nil
}

// SetMount creates or updates a mountpoint.
func (r *RadioAPI) SetMount(m *Mount) error {
	var buf bytes.Buffer
	if err := json.NewEncoder(&buf).Encode(m); err != nil {
		return err
	}

	_, err := r.client.Set(mountPath(m.Name), buf.String(), 0)
	return err
}

// DelMount removes a mountpoint.
func (r *RadioAPI) DelMount(mountName string) error {
	_, err := r.client.Delete(mountPath(mountName), false)
	return err
}

// ListMounts returns a list of all the configured mountpoints.
func (r *RadioAPI) ListMounts() ([]*Mount, error) {
	response, err := r.client.Get(MountPrefix, true, false)
	if err != nil || response.Node == nil {
		return nil, err
	}
	if !response.Node.Dir {
		return nil, ErrIsFile
	}

	result := make([]*Mount, 0, len(response.Node.Nodes))
	for _, n := range response.Node.Nodes {
		if n.Dir {
			continue
		}
		var m Mount
		if err := json.NewDecoder(strings.NewReader(n.Value)).Decode(&m); err != nil {
			continue
		}
		result = append(result, &m)
	}
	return result, nil
}

// GetMasterAddr returns the address of the current master server.
func (r *RadioAPI) GetMasterAddr() (string, error) {
	response, err := r.client.Get(MasterElectionPath, false, false)
	if err != nil || response.Node == nil {
		return "", err
	}
	if response.Node.Dir {
		return "", ErrIsDirectory
	}
	return response.Node.Value, nil
}

// GetNodes returns the list of active cluster nodes.
func (r *RadioAPI) doGetNodes() ([]*NodeStatus, error) {
	response, err := r.client.Get(NodePrefix, false, false)
	if err != nil || response.Node == nil {
		return nil, err
	}
	if !response.Node.Dir {
		return nil, ErrIsFile
	}
	result := make([]*NodeStatus, 0, len(response.Node.Nodes))
	for _, entry := range response.Node.Nodes {
		var ns NodeStatus
		if err := json.NewDecoder(strings.NewReader(entry.Value)).Decode(&ns); err == nil {
			result = append(result, &ns)
		}
	}
	return result, nil
}

func (r *RadioAPI) GetNodes() ([]*NodeStatus, error) {
	return r.activeNodesCache.Get(r.doGetNodes)
}

func (r *RadioAPI) GetNodeIPs() ([]string, error) {
	nodes, err := r.GetNodes()
	if err != nil {
		return nil, err
	}
	ips := make([]string, 0, len(nodes))
	for _, n := range nodes {
		ips = append(ips, n.IP)
	}
	return ips, nil
}

// GeneratePassword returns a new random password.
func GeneratePassword() string {
	b := make([]byte, 6)
	rand.Read(b)
	return base64.StdEncoding.EncodeToString(b)
}