<commit_before>package ostent

import (
	"fmt"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"

	gorillawebsocket "github.com/gorilla/websocket"

	"github.com/ostrost/ostent/client"
	"github.com/ostrost/ostent/flags"
)

type backgroundHandler func(flags.Period)

var (
	jobs = struct {
		mutex sync.Mutex
		added []backgroundHandler
	}{}
)

func AddBackground(j backgroundHandler) {
	jobs.mutex.Lock()
	defer jobs.mutex.Unlock()
	jobs.added = append(jobs.added, j)
}

func RunBackground(defaultPeriod flags.Period) {
	jobs.mutex.Lock()
	defer jobs.mutex.Unlock()
	for _, j := range jobs.added {
		go j(defaultPeriod)
	}
}

func init() {
	AddBackground(Loop)
}

// Loop is the ostent background job
func Loop(flags.Period) {
	go func() {
		for {
			now := time.Now()
			nextsecond := now.Truncate(time.Second).Add(time.Second).Sub(now)
			<-time.After(nextsecond)

			Connections.tick()

			if exes := Connections.expired(); len(exes) != 0 {
				lastInfo.collect(&Machine{})
				for _, c := range exes {
					c.Tack()
				}
			}
		}
	}()

	go func() {
		if err := vgwatch(); err != nil { // vagrant
			// ignoring the error
		}
	}()

	for {
		select {
		case update := <-iUPDATES:
			Connections.Push(update)

		case conn := <-Register:
			Connections.reg(conn)

		case conn := <-unregister:
			Connections.unreg(conn)
			/*
				if Connections.unreg(conn) == 0 { // if no connections left
					lastInfo.reset_prev()
				} // */
		}
	}
}

type conn struct {
	Conn *gorillawebsocket.Conn

	requestOrigin *http.Request

	receive   chan *received
	pushch    chan *IndexUpdate
	full      client.Client
	minperiod flags.Period
	access    *logger

	mutex      sync.Mutex
	writemutex sync.Mutex
}

func (c *conn) Expired() bool {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	return c.full.Expired()
}

func (c *conn) Tick() {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.full.Tick()
}

func (c *conn) Tack() {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.receive <- nil
}

func (c *conn) Push(update *IndexUpdate) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.pushch <- update
}

type receiver interface {
	Tick()
	Tack()
	Push(*IndexUpdate)
	Reload()
	Expired() bool
	CloseChans()
}

type connmap map[receiver]struct{}
type conns struct {
	connmap
	mutex sync.Mutex
}

func (cs *conns) Tack() {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()

	for c := range cs.connmap {
		c.Tack()
	}
}

// Reload sends reload signal to all the connections, returns false if there were no connections
func (cs *conns) Reload() bool {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()

	var reloaded bool
	for c := range cs.connmap {
		c.Reload()
		reloaded = true
	}
	return reloaded
}

func (cs *conns) Push(update *IndexUpdate) {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()

	for c := range cs.connmap {
		c.Push(update)
	}
}

func (cs *conns) reg(r receiver) {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()

	cs.connmap[r] = struct{}{}
}

func (c *conn) CloseChans() {
	c.mutex.Lock()
	defer func() {
		defer c.mutex.Unlock()
		if e := recover(); e != nil {
			if err, ok := e.(error); ok {
				fmt.Printf("CLOSE? PANIC %s\n", err.Error())
			} else {
				fmt.Printf("CLOSE? PANIC (UNDESCRIPT) %+v\n", e)
			}
			panic(e)
		}
	}()
	close(c.receive)
	close(c.pushch)
}

// Len return the number of active connections
func (cs *conns) Len() int {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()
	return len(cs.connmap)
}

func (cs *conns) unreg(r receiver) {
	r.CloseChans()

	cs.mutex.Lock()
	defer cs.mutex.Unlock()

	delete(cs.connmap, r)
}

func (cs *conns) tick() {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()
	for c := range cs.connmap {
		c.Tick()
	}
}

func (cs *conns) expired() []receiver {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()
	exes := []receiver{}

	for c := range cs.connmap {
		if c.Expired() {
			exes = append(exes, c)
		}
	}
	return exes
}

var (
	// Connections is an instance of unexported conns type to hold
	// active websocket connections. The only method is Reload.
	Connections = conns{connmap: make(map[receiver]struct{})}

	iUPDATES   = make(chan *IndexUpdate) // the channel for off-the-clock IndexUpdate to push
	unregister = make(chan receiver)     // (chan *conn)
	Register   = make(chan receiver)     // (chan *conn)
)

type received struct {
	Search *string
	Client *client.RecvClient
}

type served struct {
	conn     *conn // passing conn into received.ServeHTTP
	received *received
}

func (c *conn) writeJSON(data interface{}) error {
	c.writemutex.Lock()
	defer c.writemutex.Unlock()
	return c.Conn.WriteJSON(data)
}

func (c *conn) Reload() {
	c.writeJSON(struct {<commit_msg>buffered Register channel<commit_after>package ostent

import (
	"fmt"
	"net/http"
	"net/url"
	"strings"
	"sync"
	"time"

	gorillawebsocket "github.com/gorilla/websocket"

	"github.com/ostrost/ostent/client"
	"github.com/ostrost/ostent/flags"
)

type backgroundHandler func(flags.Period)

var (
	jobs = struct {
		mutex sync.Mutex
		added []backgroundHandler
	}{}
)

func AddBackground(j backgroundHandler) {
	jobs.mutex.Lock()
	defer jobs.mutex.Unlock()
	jobs.added = append(jobs.added, j)
}

func RunBackground(defaultPeriod flags.Period) {
	jobs.mutex.Lock()
	defer jobs.mutex.Unlock()
	for _, j := range jobs.added {
		go j(defaultPeriod)
	}
}

func init() {
	AddBackground(Loop)
}

// SleepTilNextSecond sleeps til precisely next second.
func SleepTilNextSecond() {
	now := time.Now()
	nextsecond := now.Truncate(time.Second).Add(time.Second).Sub(now)
	time.Sleep(nextsecond)
}

// Loop is the ostent background job
func Loop(flags.Period) {
	go func() {
		for {
			SleepTilNextSecond()

			Connections.tick()

			if exes := Connections.expired(); len(exes) != 0 {
				lastInfo.collect(&Machine{})
				for _, c := range exes {
					c.Tack()
				}
			}
		}
	}()

	go func() {
		if err := vgwatch(); err != nil { // vagrant
			// ignoring the error
		}
	}()

	for {
		select {
		case update := <-iUPDATES:
			Connections.Push(update)

		case conn := <-Register:
			Connections.reg(conn)

		case conn := <-unregister:
			Connections.unreg(conn)
			/*
				if Connections.unreg(conn) == 0 { // if no connections left
					lastInfo.reset_prev()
				} // */
		}
	}
}

type conn struct {
	Conn *gorillawebsocket.Conn

	requestOrigin *http.Request

	receive   chan *received
	pushch    chan *IndexUpdate
	full      client.Client
	minperiod flags.Period
	access    *logger

	mutex      sync.Mutex
	writemutex sync.Mutex
}

func (c *conn) Expired() bool {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	return c.full.Expired()
}

func (c *conn) Tick() {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.full.Tick()
}

func (c *conn) Tack() {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.receive <- nil
}

func (c *conn) Push(update *IndexUpdate) {
	c.mutex.Lock()
	defer c.mutex.Unlock()
	c.pushch <- update
}

type receiver interface {
	Tick()
	Tack()
	Push(*IndexUpdate)
	Reload()
	Expired() bool
	CloseChans()
}

type connmap map[receiver]struct{}
type conns struct {
	connmap
	mutex sync.Mutex
}

func (cs *conns) Tack() {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()

	for c := range cs.connmap {
		c.Tack()
	}
}

// Reload sends reload signal to all the connections, returns false if there were no connections
func (cs *conns) Reload() bool {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()

	var reloaded bool
	for c := range cs.connmap {
		c.Reload()
		reloaded = true
	}
	return reloaded
}

func (cs *conns) Push(update *IndexUpdate) {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()

	for c := range cs.connmap {
		c.Push(update)
	}
}

func (cs *conns) reg(r receiver) {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()

	cs.connmap[r] = struct{}{}
}

func (c *conn) CloseChans() {
	c.mutex.Lock()
	defer func() {
		defer c.mutex.Unlock()
		if e := recover(); e != nil {
			if err, ok := e.(error); ok {
				fmt.Printf("CLOSE? PANIC %s\n", err.Error())
			} else {
				fmt.Printf("CLOSE? PANIC (UNDESCRIPT) %+v\n", e)
			}
			panic(e)
		}
	}()
	close(c.receive)
	close(c.pushch)
}

// Len return the number of active connections
func (cs *conns) Len() int {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()
	return len(cs.connmap)
}

func (cs *conns) unreg(r receiver) {
	r.CloseChans()

	cs.mutex.Lock()
	defer cs.mutex.Unlock()

	delete(cs.connmap, r)
}

func (cs *conns) tick() {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()
	for c := range cs.connmap {
		c.Tick()
	}
}

func (cs *conns) expired() []receiver {
	cs.mutex.Lock()
	defer cs.mutex.Unlock()
	exes := []receiver{}

	for c := range cs.connmap {
		if c.Expired() {
			exes = append(exes, c)
		}
	}
	return exes
}

var (
	// Connections is an instance of unexported conns type to hold
	// active websocket connections. The only method is Reload.
	Connections = conns{connmap: make(map[receiver]struct{})}

	iUPDATES   = make(chan *IndexUpdate) // the channel for off-the-clock IndexUpdate to push
	unregister = make(chan receiver)
	Register   = make(chan receiver, 1)
)

type received struct {
	Search *string
	Client *client.RecvClient
}

type served struct {
	conn     *conn // passing conn into received.ServeHTTP
	received *received
}

func (c *conn) writeJSON(data interface{}) error {
	c.writemutex.Lock()
	defer c.writemutex.Unlock()
	return c.Conn.WriteJSON(data)
}

func (c *conn) Reload() {
	c.writeJSON(struct {
{'fraction_non_alphanumeric': 0.1501521077792264, 'fraction_numerical': 0.0005432420686657974, 'mean_word_length': 3.3215962441314555, 'pattern_counts': {'":': 0, '<': 14, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '7504031', 'n_tokens_mistral': 3731, 'n_tokens_neox': 3477, 'n_words': 860}