114 lines
3.5 KiB
Go
114 lines
3.5 KiB
Go
package service
|
|
|
|
import (
|
|
"browser.local/platform/domain"
|
|
"browser.local/platform/validator"
|
|
)
|
|
|
|
const defaultLogQueryLimit = 100
|
|
|
|
func (svc *CoreService) IngestLogBatch(batch domain.LogBatchIngest) (domain.LogBatchIngestResult, error) {
|
|
batch = domain.CopyLogBatchIngest(batch)
|
|
if err := validator.ValidateLogBatchIngest(batch); err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
if err := svc.validateRunSession(batch.RunEndpointID, batch.SessionToken); err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
|
|
stamp := svc.now()
|
|
stream, err := svc.store.LogStreams().Get(batch.LogStreamID)
|
|
if err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
if err := validateLogBatchStream(batch, stream); err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
|
|
if batch.LastSeq <= stream.LatestSeq {
|
|
record, exists, err := svc.logStore.GetBatch(batch.LogStreamID, batch.FirstSeq)
|
|
if err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
if exists && record.LastSeq == batch.LastSeq && record.Checksum == batch.Checksum {
|
|
return domain.LogBatchIngestResult{
|
|
Accepted: true,
|
|
LogStreamID: batch.LogStreamID,
|
|
AcceptedFrom: batch.FirstSeq,
|
|
AcceptedTo: batch.LastSeq,
|
|
LatestSeq: stream.LatestSeq,
|
|
Duplicate: true,
|
|
ServerTime: stamp,
|
|
}, nil
|
|
}
|
|
return domain.LogBatchIngestResult{}, validationError("log batch conflicts with acknowledged range")
|
|
}
|
|
if batch.FirstSeq != stream.LatestSeq+1 {
|
|
return domain.LogBatchIngestResult{}, validationError("log batch firstSeq must follow latest acknowledged sequence")
|
|
}
|
|
|
|
record := domain.CopyLogBatchRecord(domain.LogBatchRecord{
|
|
Checksum: batch.Checksum,
|
|
FirstSeq: batch.FirstSeq,
|
|
LastSeq: batch.LastSeq,
|
|
Entries: batch.Entries,
|
|
})
|
|
if err := svc.logStore.AppendBatch(batch.LogStreamID, record); err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
stream.LatestSeq = batch.LastSeq
|
|
stream.UpdatedAt = stamp
|
|
if err := svc.store.LogStreams().Update(stream); err != nil {
|
|
return domain.LogBatchIngestResult{}, err
|
|
}
|
|
return domain.LogBatchIngestResult{
|
|
Accepted: true,
|
|
LogStreamID: batch.LogStreamID,
|
|
AcceptedFrom: batch.FirstSeq,
|
|
AcceptedTo: batch.LastSeq,
|
|
LatestSeq: stream.LatestSeq,
|
|
ServerTime: stamp,
|
|
}, nil
|
|
}
|
|
|
|
func (svc *CoreService) QueryLogStream(query domain.LogStreamCursorQuery) (domain.LogStreamCursorResult, error) {
|
|
if err := validator.ValidateLogStreamCursorQuery(query); err != nil {
|
|
return domain.LogStreamCursorResult{}, err
|
|
}
|
|
stream, err := svc.store.LogStreams().Get(query.LogStreamID)
|
|
if err != nil {
|
|
return domain.LogStreamCursorResult{}, err
|
|
}
|
|
limit := query.Limit
|
|
if limit == 0 {
|
|
limit = defaultLogQueryLimit
|
|
}
|
|
|
|
selected, nextSeq, err := svc.logStore.Query(query.LogStreamID, query.AfterSeq, limit)
|
|
if err != nil {
|
|
return domain.LogStreamCursorResult{}, err
|
|
}
|
|
return domain.CopyLogStreamCursorResult(domain.LogStreamCursorResult{
|
|
LogStreamID: query.LogStreamID,
|
|
Entries: selected,
|
|
NextSeq: nextSeq,
|
|
LatestSeq: stream.LatestSeq,
|
|
}), nil
|
|
}
|
|
|
|
func validateLogBatchStream(batch domain.LogBatchIngest, stream domain.LogStream) error {
|
|
if stream.ID != batch.LogStreamID {
|
|
return validationError("logStreamId must match stream")
|
|
}
|
|
if stream.ServerInstanceID != batch.ServerInstanceID {
|
|
return validationError("serverInstanceId must match stream")
|
|
}
|
|
if stream.StreamKey != batch.StreamKey {
|
|
return validationError("streamKey must match stream")
|
|
}
|
|
if stream.Source != batch.Source {
|
|
return validationError("source must match stream")
|
|
}
|
|
return nil
|
|
}
|