browser/platform/dto/log_ingest.go

package dto

import (
	"time"

	"browser.local/platform/domain"
)

type LogEntryBody struct {
	Seq       uint64            `json:"seq"`
	Timestamp time.Time         `json:"timestamp"`
	Level     string            `json:"level,omitempty"`
	Line      string            `json:"line"`
	Fields    map[string]string `json:"fields,omitempty"`
	Redacted  bool              `json:"redacted"`
}

type LogBatchIngestRequest struct {
	RunEndpointID    string                 `json:"runEndpointId"`
	SessionToken     string                 `json:"sessionToken"`
	LogStreamID      string                 `json:"logStreamId"`
	ServerInstanceID string                 `json:"serverInstanceId"`
	StreamKey        string                 `json:"streamKey"`
	Source           domain.LogStreamSource `json:"source"`
	FirstSeq         uint64                 `json:"firstSeq"`
	LastSeq          uint64                 `json:"lastSeq"`
	Compression      string                 `json:"compression"`
	Checksum         string                 `json:"checksum"`
	Entries          []LogEntryBody         `json:"entries"`
}

type LogBatchIngestResponse struct {
	Accepted      bool      `json:"accepted"`
	LogStreamID   string    `json:"logStreamId"`
	AcceptedFrom  uint64    `json:"acceptedFrom"`
	AcceptedTo    uint64    `json:"acceptedTo"`
	LatestSeq     uint64    `json:"latestSeq"`
	Duplicate     bool      `json:"duplicate"`
	RetryAfterSec int       `json:"retryAfterSec,omitempty"`
	ServerTime    time.Time `json:"serverTime"`
}

type LogStreamCursorRequest struct {
	LogStreamID string `json:"logStreamId"`
	AfterSeq    uint64 `json:"afterSeq"`
	Limit       int    `json:"limit"`
}

type LogStreamCursorResponse struct {
	LogStreamID string         `json:"logStreamId"`
	Entries     []LogEntryBody `json:"entries"`
	NextSeq     uint64         `json:"nextSeq"`
	LatestSeq   uint64         `json:"latestSeq"`
}

type RunLogStreamProgressRequest struct {
	RunEndpointID    string `json:"runEndpointId"`
	SessionToken     string `json:"sessionToken"`
	ServerInstanceID string `json:"serverInstanceId"`
	LogStreamID      string `json:"logStreamId"`
}

type RunLogStreamProgressResponse struct {
	Accepted         bool      `json:"accepted"`
	RunEndpointID    string    `json:"runEndpointId"`
	ServerInstanceID string    `json:"serverInstanceId"`
	LogStreamID      string    `json:"logStreamId"`
	LatestSeq        uint64    `json:"latestSeq"`
	ServerTime       time.Time `json:"serverTime"`
}

type LogStreamEventResponse struct {
	ServerInstanceID string                 `json:"serverInstanceId"`
	StreamID         string                 `json:"streamId"`
	Source           domain.LogStreamSource `json:"source"`
	StreamKey        string                 `json:"streamKey"`
	LatestSeq        uint64                 `json:"latestSeq"`
	Entry            LogEntryBody           `json:"entry"`
}

type LogStreamEventsReadyResponse struct {
	ServerInstanceID string    `json:"serverInstanceId"`
	StreamCount      int       `json:"streamCount"`
	ServerTime       time.Time `json:"serverTime"`
}

func (request LogBatchIngestRequest) ToDomain() domain.LogBatchIngest {
	return domain.LogBatchIngest{
		RunEndpointID:    request.RunEndpointID,
		SessionToken:     request.SessionToken,
		LogStreamID:      request.LogStreamID,
		ServerInstanceID: request.ServerInstanceID,
		StreamKey:        request.StreamKey,
		Source:           request.Source,
		FirstSeq:         request.FirstSeq,
		LastSeq:          request.LastSeq,
		Compression:      request.Compression,
		Checksum:         request.Checksum,
		Entries:          logEntriesToDomain(request.Entries),
	}
}

func (request LogStreamCursorRequest) ToDomain() domain.LogStreamCursorQuery {
	return domain.LogStreamCursorQuery{
		LogStreamID: request.LogStreamID,
		AfterSeq:    request.AfterSeq,
		Limit:       request.Limit,
	}
}

func (request RunLogStreamProgressRequest) ToDomain() domain.RunLogStreamProgress {
	return domain.RunLogStreamProgress{RunEndpointID: request.RunEndpointID, SessionToken: request.SessionToken, ServerInstanceID: request.ServerInstanceID, LogStreamID: request.LogStreamID}
}

func LogBatchIngestFromDomain(result domain.LogBatchIngestResult) LogBatchIngestResponse {
	return LogBatchIngestResponse{
		Accepted:     result.Accepted,
		LogStreamID:  result.LogStreamID,
		AcceptedFrom: result.AcceptedFrom,
		AcceptedTo:   result.AcceptedTo,
		LatestSeq:    result.LatestSeq,
		Duplicate:    result.Duplicate,
		ServerTime:   result.ServerTime,
	}
}

func LogStreamCursorFromDomain(result domain.LogStreamCursorResult) LogStreamCursorResponse {
	result = domain.CopyLogStreamCursorResult(result)
	return LogStreamCursorResponse{
		LogStreamID: result.LogStreamID,
		Entries:     logEntriesFromDomain(result.Entries),
		NextSeq:     result.NextSeq,
		LatestSeq:   result.LatestSeq,
	}
}

func RunLogStreamProgressFromDomain(result domain.RunLogStreamProgressResult) RunLogStreamProgressResponse {
	return RunLogStreamProgressResponse{Accepted: result.Accepted, RunEndpointID: result.RunEndpointID, ServerInstanceID: result.ServerInstanceID, LogStreamID: result.LogStreamID, LatestSeq: result.LatestSeq, ServerTime: result.ServerTime}
}

func LogStreamEventFromDomain(event domain.LogStreamEvent) LogStreamEventResponse {
	event = domain.CopyLogStreamEvent(event)
	return LogStreamEventResponse{
		ServerInstanceID: event.ServerInstanceID,
		StreamID:         event.Stream.ID,
		Source:           event.Stream.Source,
		StreamKey:        event.Stream.StreamKey,
		LatestSeq:        event.LatestSeq,
		Entry:            logEntryFromDomain(event.Entry),
	}
}

func logEntriesToDomain(entries []LogEntryBody) []domain.LogEntry {
	if entries == nil {
		return nil
	}
	out := make([]domain.LogEntry, len(entries))
	for i, entry := range entries {
		out[i] = domain.LogEntry{
			Seq:       entry.Seq,
			Timestamp: entry.Timestamp,
			Level:     entry.Level,
			Line:      entry.Line,
			Fields:    copyStringMap(entry.Fields),
			Redacted:  entry.Redacted,
		}
	}
	return out
}

func logEntryFromDomain(entry domain.LogEntry) LogEntryBody {
	return LogEntryBody{
		Seq:       entry.Seq,
		Timestamp: entry.Timestamp,
		Level:     entry.Level,
		Line:      entry.Line,
		Fields:    copyStringMap(entry.Fields),
		Redacted:  entry.Redacted,
	}
}

func logEntriesFromDomain(entries []domain.LogEntry) []LogEntryBody {
	if entries == nil {
		return nil
	}
	out := make([]LogEntryBody, len(entries))
	for i, entry := range entries {
		out[i] = logEntryFromDomain(entry)
	}
	return out
}

func copyStringMap(values map[string]string) map[string]string {
	if values == nil {
		return nil
	}
	out := make(map[string]string, len(values))
	for key, value := range values {
		out[key] = value
	}
	return out
}