<commit_before>package builder

import (
	"strings"
	"sync"

	"github.com/velocity-ci/velocity/backend/pkg/domain/build"
	"github.com/velocity-ci/velocity/backend/pkg/domain/builder"
	"go.uber.org/zap"

	"github.com/gorilla/websocket"
	"github.com/velocity-ci/velocity/backend/pkg/velocity"
)

type safeWebsocket struct {
	ws   *websocket.Conn
	lock sync.RWMutex
}

func (sws *safeWebsocket) WriteJSON(m interface{}) error {
	sws.lock.Lock()
	defer sws.lock.Unlock()

	return sws.ws.WriteJSON(m)
}

type StreamWriter struct {
	ws         *safeWebsocket
	StepNumber int

	BuildID  string
	StepID   string
	StreamID string

	LineNumber int
	status     string
}

type Emitter struct {
	ws      *safeWebsocket
	BuildID string
	StepID  string
	Streams []*build.Stream

	StepNumber int
}

func (e *Emitter) GetStreamWriter(streamName string) velocity.StreamWriter {
	streamID := ""
	for _, s := range e.Streams {
		if s.Name == streamName {
			streamID = s.ID
			break
		}
	}
	if streamID == "" {
		velocity.GetLogger().Error("could not find streamID", zap.String("stream name", streamName))
	}
	return &StreamWriter{
		ws:         e.ws,
		BuildID:    e.BuildID,
		StepID:     e.StepID,
		StreamID:   streamID,
		StepNumber: e.StepNumber,
		LineNumber: 0,
	}
}

func (e *Emitter) SetStepAndStreams(step *build.Step, streams []*build.Stream) {
	e.StepID = step.ID
	e.Streams = []*build.Stream{}
	for _, s := range streams {
		if s.Step.ID == step.ID {
			e.Streams = append(e.Streams, s)
		}
	}
}

func (e *Emitter) SetStepNumber(n int) {
	e.StepNumber = n
}

func NewEmitter(ws *websocket.Conn, b *build.Build) *Emitter {
	return &Emitter{
		ws:      &safeWebsocket{ws: ws},
		BuildID: b.ID,
	}
}

func (w *StreamWriter) Write(p []byte) (n int, err error) {
	o := strings.TrimSpace(string(p))
	if !strings.Contains(string(p), "\r") {
		w.LineNumber++
		o += "\n"
	} else {
		parts := strings.Split(o, "\r")
		o = parts[len(parts)-1] + "\r"
	}

	lM := builder.BuilderStreamLineMessage{
		BuildID:    w.BuildID,
		StepID:     w.StepID,
		StreamID:   w.StreamID,
		LineNumber: w.LineNumber,
		Status:     w.status,
		Output:     o,
	}
	m := builder.BuilderRespMessage{
		Type: "log",
		Data: lM,
	}
	err = w.ws.WriteJSON(m)

	if err != nil {
		return 0, err
	}

	return len(p), nil
}

func (w *StreamWriter) SetStatus(s string) {
	w.status = s
}<commit_msg>Less complicated handling of carriage returns<commit_after>package builder

import (
	"strings"
	"sync"

	"github.com/velocity-ci/velocity/backend/pkg/domain/build"
	"github.com/velocity-ci/velocity/backend/pkg/domain/builder"
	"go.uber.org/zap"

	"github.com/gorilla/websocket"
	"github.com/velocity-ci/velocity/backend/pkg/velocity"
)

type safeWebsocket struct {
	ws   *websocket.Conn
	lock sync.RWMutex
}

func (sws *safeWebsocket) WriteJSON(m interface{}) error {
	sws.lock.Lock()
	defer sws.lock.Unlock()

	return sws.ws.WriteJSON(m)
}

type StreamWriter struct {
	ws         *safeWebsocket
	StepNumber int

	BuildID  string
	StepID   string
	StreamID string

	LineNumber int
	status     string
}

type Emitter struct {
	ws      *safeWebsocket
	BuildID string
	StepID  string
	Streams []*build.Stream

	StepNumber int
}

func (e *Emitter) GetStreamWriter(streamName string) velocity.StreamWriter {
	streamID := ""
	for _, s := range e.Streams {
		if s.Name == streamName {
			streamID = s.ID
			break
		}
	}
	if streamID == "" {
		velocity.GetLogger().Error("could not find streamID", zap.String("stream name", streamName))
	}
	return &StreamWriter{
		ws:         e.ws,
		BuildID:    e.BuildID,
		StepID:     e.StepID,
		StreamID:   streamID,
		StepNumber: e.StepNumber,
		LineNumber: 0,
	}
}

func (e *Emitter) SetStepAndStreams(step *build.Step, streams []*build.Stream) {
	e.StepID = step.ID
	e.Streams = []*build.Stream{}
	for _, s := range streams {
		if s.Step.ID == step.ID {
			e.Streams = append(e.Streams, s)
		}
	}
}

func (e *Emitter) SetStepNumber(n int) {
	e.StepNumber = n
}

func NewEmitter(ws *websocket.Conn, b *build.Build) *Emitter {
	return &Emitter{
		ws:      &safeWebsocket{ws: ws},
		BuildID: b.ID,
	}
}

func (w *StreamWriter) Write(p []byte) (n int, err error) {
	o := strings.TrimSpace(string(p))
	if !strings.HasSuffix(string(p), "\r") {
		w.LineNumber++
		o += "\n"
	}

	lM := builder.BuilderStreamLineMessage{
		BuildID:    w.BuildID,
		StepID:     w.StepID,
		StreamID:   w.StreamID,
		LineNumber: w.LineNumber,
		Status:     w.status,
		Output:     o,
	}
	m := builder.BuilderRespMessage{
		Type: "log",
		Data: lM,
	}
	err = w.ws.WriteJSON(m)

	if err != nil {
		return 0, err
	}

	return len(p), nil
}

func (w *StreamWriter) SetStatus(s string) {
	w.status = s
}
{'fraction_non_alphanumeric': 0.14079965606190886, 'fraction_numerical': 0.0010748065348237317, 'mean_word_length': 3.3364398881640263, 'pattern_counts': {'":': 0, '<': 3, '<?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': '4095444', 'n_tokens_mistral': 1888, 'n_tokens_neox': 1800, 'n_words': 400}