<commit_before>// Package observer provides a facility for publishing progress updates and
// state changes from parts of the daemon, an a SSE http handler for consumers
// of these events.
package observer

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
)

type ctxkey string

// CtxRequestID is the context WithValue key for a request id.
var CtxRequestID ctxkey = "id"

// EventType represents all possible types of events that can be observed.
type EventType string

// All values for EventType
const (
	Progress EventType = "progress"
	Started  EventType = "started"
	Finished EventType = "finished"
	Errored  EventType = "errored"
	Aborted  EventType = "aborted"
)

type event struct {
	ID        string    `json:"id"`
	Type      EventType `json:"-"` // type is included in the SSE
	Message   string    `json:"message"`
	Completed uint      `json:"completed"`
	Total     uint      `json:"total"`
}

type notification struct {
	Type      EventType
	Message   string
	Increment bool
}

// Observer receives events via Notify, and publishes them as SSEs via its
// ServeHTTP function.
type Observer struct {
	notify chan *event
	closed chan int

	observers map[chan []byte]bool

	newObservers    chan chan []byte
	closedObservers chan chan []byte
}

type transaction struct {
	requestID      string
	total          uint
	current        uint
	events         chan<- *event
	totalUpdates   <-chan uint
	notifications  <-chan *notification
	observerClosed <-chan int
	ctxDone        <-chan struct{}
}

// Notifier belongs to a transactions and represents one segment in a series of
// actions. A Notifier can send many messages.
type Notifier struct {
	total          uint
	current        uint
	transaction    *transaction
	totalUpdates   chan<- uint
	notifications  chan<- *notification
	observerClosed <-chan int
	ctxDone        <-chan struct{}
}

func (t *transaction) start() {
	go func() {
		for {
			select {
			case notification := <-t.notifications:
				if notification.Increment {
					t.current++
				}

				evt := &event{
					ID:        t.requestID,
					Type:      notification.Type,
					Message:   notification.Message,
					Completed: t.current,
					Total:     t.total,
				}

				t.events <- evt
			case size := <-t.totalUpdates:
				t.total += size
			case <-t.ctxDone: // transaction has completed
				return
			case <-t.observerClosed: // observer has shutdown
				return
			}
		}
	}()
}

// Notifier creates a child notifier to this Notifier
func (n *Notifier) Notifier(total uint) *Notifier {
	notifier := &Notifier{
		total:          total,
		current:        0,
		transaction:    nil,
		totalUpdates:   n.totalUpdates,
		notifications:  n.notifications,
		observerClosed: n.observerClosed,
		ctxDone:        n.ctxDone,
	}

	n.totalUpdates <- total
	return notifier
}

// Notify publishes an event to all SSE observers.
func (n *Notifier) Notify(eventType EventType, message string, increment bool) {
	notif := &notification{
		Type:      eventType,
		Message:   message,
		Increment: increment,
	}

	if increment {
		n.current++
	}

	if n.current > n.total {
		panic(fmt.Sprintf(
			"notifications exceed maximum %d/%d", n.current, n.total))
	}

	select {
	case n.notifications <- notif:
		return
	case <-n.observerClosed:
		return
	case <-n.ctxDone:
		return
	}
}

// New returns a new initialized Observer
func New() *Observer {
	return &Observer{<commit_msg>Observer: put a Read/Write mutex in place.<commit_after>// Package observer provides a facility for publishing progress updates and
// state changes from parts of the daemon, an a SSE http handler for consumers
// of these events.
package observer

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"net/http"
	"sync"
)

type ctxkey string

// CtxRequestID is the context WithValue key for a request id.
var CtxRequestID ctxkey = "id"

// EventType represents all possible types of events that can be observed.
type EventType string

// All values for EventType
const (
	Progress EventType = "progress"
	Started  EventType = "started"
	Finished EventType = "finished"
	Errored  EventType = "errored"
	Aborted  EventType = "aborted"
)

type event struct {
	ID        string    `json:"id"`
	Type      EventType `json:"-"` // type is included in the SSE
	Message   string    `json:"message"`
	Completed uint      `json:"completed"`
	Total     uint      `json:"total"`
}

type notification struct {
	Type      EventType
	Message   string
	Increment bool
}

// Observer receives events via Notify, and publishes them as SSEs via its
// ServeHTTP function.
type Observer struct {
	notify chan *event
	closed chan int

	observers map[chan []byte]bool

	newObservers    chan chan []byte
	closedObservers chan chan []byte
}

type transaction struct {
	requestID      string
	total          uint
	current        uint
	events         chan<- *event
	totalUpdates   <-chan uint
	notifications  <-chan *notification
	observerClosed <-chan int
	ctxDone        <-chan struct{}
}

// Notifier belongs to a transactions and represents one segment in a series of
// actions. A Notifier can send many messages.
type Notifier struct {
	total          uint
	current        uint
	transaction    *transaction
	totalUpdates   chan<- uint
	notifications  chan<- *notification
	observerClosed <-chan int
	ctxDone        <-chan struct{}
	sync.RWMutex
}

func (t *transaction) start() {
	go func() {
		for {
			select {
			case notification := <-t.notifications:
				if notification.Increment {
					t.current++
				}

				evt := &event{
					ID:        t.requestID,
					Type:      notification.Type,
					Message:   notification.Message,
					Completed: t.current,
					Total:     t.total,
				}

				t.events <- evt
			case size := <-t.totalUpdates:
				t.total += size
			case <-t.ctxDone: // transaction has completed
				return
			case <-t.observerClosed: // observer has shutdown
				return
			}
		}
	}()
}

// Notifier creates a child notifier to this Notifier
func (n *Notifier) Notifier(total uint) *Notifier {
	notifier := &Notifier{
		total:          total,
		current:        0,
		transaction:    nil,
		totalUpdates:   n.totalUpdates,
		notifications:  n.notifications,
		observerClosed: n.observerClosed,
		ctxDone:        n.ctxDone,
	}

	n.totalUpdates <- total
	return notifier
}

// Notify publishes an event to all SSE observers.
func (n *Notifier) Notify(eventType EventType, message string, increment bool) {
	notif := &notification{
		Type:      eventType,
		Message:   message,
		Increment: increment,
	}

	if increment {
		n.Lock()
		n.current++
		n.Unlock()
	}

	n.RLock()
	if n.current > n.total {
		panic(fmt.Sprintf(
			"notifications exceed maximum %d/%d", n.current, n.total))
	}
	n.RUnlock()

	select {
	case n.notifications <- notif:
		return
	case <-n.observerClosed:
		return
	case <-n.ctxDone:
		return
	}
}

// New returns a new initialized Observer
func New() *Observer {
	return &Observer{
{'fraction_non_alphanumeric': 0.09452232981644747, 'fraction_numerical': 0.00028905911258852437, 'mean_word_length': 2.8702460850111855, 'pattern_counts': {'":': 0, '<': 39, '<?xml version=': 0, '>': 5, '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': '5257110', 'n_tokens_mistral': 2390, 'n_tokens_neox': 2123, 'n_words': 729}