diff --git a/apps/container-registry/internal/domain/build-run.go b/apps/container-registry/internal/domain/build-run.go index a6e47a352..0718ca5f0 100644 --- a/apps/container-registry/internal/domain/build-run.go +++ b/apps/container-registry/internal/domain/build-run.go @@ -3,7 +3,6 @@ package domain import ( "crypto/md5" "fmt" - "strconv" "strings" "time" @@ -52,32 +51,32 @@ func (d *Impl) GetBuildRun(ctx RegistryContext, repoName string, buildRunName st return brun, nil } -func (d *Impl) parseRecordVersionFromAnnotations(annotations map[string]string) (int, error) { - annotatedVersion, ok := annotations[constants.RecordVersionKey] - if !ok { - return 0, errors.Newf("no annotation with record version key (%s), found on the resource", constants.RecordVersionKey) - } - - annVersion, err := strconv.ParseInt(annotatedVersion, 10, 32) - if err != nil { - return 0, errors.NewE(err) - } +// func (d *Impl) parseRecordVersionFromAnnotations(annotations map[string]string) (int, error) { +// annotatedVersion, ok := annotations[constants.RecordVersionKey] +// if !ok { +// return 0, errors.Newf("no annotation with record version key (%s), found on the resource", constants.RecordVersionKey) +// } +// +// annVersion, err := strconv.ParseInt(annotatedVersion, 10, 32) +// if err != nil { +// return 0, errors.NewE(err) +// } +// +// return int(annVersion), nil +// } - return int(annVersion), nil -} - -func (d *Impl) MatchRecordVersion(annotations map[string]string, rv int) (int, error) { - annVersion, err := d.parseRecordVersionFromAnnotations(annotations) - if err != nil { - return -1, errors.NewE(err) - } - - if annVersion != rv { - return -1, errors.Newf("record version mismatch, expected %d, got %d", rv, annVersion) - } - - return annVersion, nil -} +// func (d *Impl) MatchRecordVersion(annotations map[string]string, rv int) (int, error) { +// annVersion, err := d.parseRecordVersionFromAnnotations(annotations) +// if err != nil { +// return -1, errors.NewE(err) +// } +// +// if annVersion != rv { +// return -1, errors.Newf("record version mismatch, expected %d, got %d", rv, annVersion) +// } +// +// return annVersion, nil +// } func (d *Impl) OnBuildRunUpdateMessage(ctx RegistryContext, buildRun entities.BuildRun, status t2.ResourceStatus, opts UpdateAndDeleteOpts) error { @@ -94,15 +93,15 @@ func (d *Impl) OnBuildRunUpdateMessage(ctx RegistryContext, buildRun entities.Bu return errors.Newf("build run with name %q not found", buildRun.Name) } - recordVersion, err := d.MatchRecordVersion(xBr.Annotations, xBr.RecordVersion) - if err != nil { - return errors.NewE(err) - } + // recordVersion, err := d.MatchRecordVersion(xBr.Annotations, xBr.RecordVersion) + // if err != nil { + // return errors.NewE(err) + // } if _, err = d.buildRunRepo.PatchById( ctx, xBr.Id, - common.PatchForSyncFromAgent(&buildRun, recordVersion, status, common.PatchOpts{ + common.PatchForSyncFromAgent(&buildRun, 0, status, common.PatchOpts{ MessageTimestamp: opts.MessageTimestamp, })); err != nil { return errors.NewE(err) diff --git a/apps/websocket-server/internal/domain/logs.go b/apps/websocket-server/internal/domain/logs.go index fc22ed859..3092ca4be 100644 --- a/apps/websocket-server/internal/domain/logs.go +++ b/apps/websocket-server/internal/domain/logs.go @@ -91,7 +91,10 @@ func (d *domain) handleLogsMsg(ctx types.Context, logsSubs *logs.LogsSubsMap, ms } } - jc, err := d.newJetstreamConsumerForLog(ctx.Context, logs.LogSubsId(msg.Spec, d.env.LogsStreamName), hash, msg.Spec.Since) + tpk := logs.LogSubsId(msg.Spec, d.env.LogsStreamName) + d.logger.Debugf("tpk: %s", tpk) + + jc, err := d.newJetstreamConsumerForLog(ctx.Context, tpk, hash, msg.Spec.Since) if err != nil { return err } diff --git a/apps/websocket-server/internal/domain/logs/main.go b/apps/websocket-server/internal/domain/logs/main.go index fdbff3b8e..03a579922 100644 --- a/apps/websocket-server/internal/domain/logs/main.go +++ b/apps/websocket-server/internal/domain/logs/main.go @@ -106,5 +106,5 @@ func LogSubsId(md MsgData, logStreamName string) string { return fmt.Sprintf("%s.%s.%s.%s.>", logStreamName, md.Account, md.Cluster, md.TrackingId) } - return fmt.Sprintf("%s.%s.%s.%s.%d.>", logStreamName, md.Account, md.Cluster, md.TrackingId, md.RecordVersion) + return fmt.Sprintf("%s.%s.%s.%s.%d.>", logStreamName, md.Account, md.Cluster, md.TrackingId, *md.RecordVersion) }