package main

import (
	"sync"

	"code.justin.tv/release/trace/api"
	"code.justin.tv/release/trace/pbmsg"
)

type query struct {
	pbCh chan *pbmsg.Transaction
	ch   chan *api.Transaction

	req *api.FirehoseRequest
}

type queryManager struct {
	queries []*query
	lock    *sync.RWMutex
}

func newQueryManager() *queryManager {
	return &queryManager{
		queries: make([]*query, 0),
		lock:    &sync.RWMutex{},
	}
}

func (m *queryManager) Add(req *api.FirehoseRequest) (<-chan *api.Transaction, <-chan *pbmsg.Transaction) {
	m.lock.Lock()
	defer m.lock.Unlock()
	ch := make(chan *api.Transaction, 64)
	pbCh := make(chan *pbmsg.Transaction, 64)
	m.queries = append(m.queries, &query{
		req:  req,
		ch:   ch,
		pbCh: pbCh,
	})
	return ch, pbCh
}

func (m *queryManager) Remove(req *api.FirehoseRequest) {
	m.lock.Lock()
	defer m.lock.Unlock()
	for index, query := range m.queries {
		if query.req == req {
			m.queries = append(m.queries[:index], m.queries[index+1:]...)
			close(query.ch)
			close(query.pbCh)
		}
	}
}

func (m *queryManager) Each(f func(*query)) {
	m.lock.RLock()
	defer m.lock.RUnlock()
	for _, query := range m.queries {
		f(query)
	}
}
