Implement platform management features

platform/service/resources.go

@@ -233,6 +233,7 @@ type CoreService struct {
 	bridgeMu              sync.Mutex
 	bridgeSeq             uint64
 	logStore              LogBodyStore
+	logIngestMu           [64]sync.Mutex
 	logEventMu            sync.Mutex
 	logEventSubscribers   map[uint64]logEventSubscriber
 	logEventSubscriberSeq uint64
@@ -2476,6 +2477,13 @@ func (svc *CoreService) CreateJob(job domain.Job) (domain.Job, error) {
 }
 
 func (svc *CoreService) ensureJobLogStreams(job domain.Job, stamp time.Time) error {
+	lock := svc.logIngestLock(job.ServerInstanceID)
+	lock.Lock()
+	defer lock.Unlock()
+	return svc.ensureJobLogStreamsUnlocked(job, stamp)
+}
+
+func (svc *CoreService) ensureJobLogStreamsUnlocked(job domain.Job, stamp time.Time) error {
 	if strings.TrimSpace(job.ServerInstanceID) == "" || strings.TrimSpace(job.ID) == "" {
 		return nil
 	}
@@ -2524,7 +2532,7 @@ func (svc *CoreService) ensureJobLogStreams(job domain.Job, stamp time.Time) err
 			CreatedAt:        stamp,
 			UpdatedAt:        stamp,
 		}
-		if _, err := svc.CreateLogStream(stream); err != nil && !errors.Is(err, repo.ErrDuplicate) {
+		if _, err := svc.createLogStream(stream); err != nil && !errors.Is(err, repo.ErrDuplicate) {
 			return err
 		}
 	}
@@ -2539,6 +2547,10 @@ func runLogStreamID(runEndpointID string, serverInstanceID string, streamKey str
 	return fmt.Sprintf("run.%s.%s.%s", runEndpointID, serverInstanceID, streamKey)
 }
 
+func runSessionLogStreamID(runEndpointID string, serverInstanceID string, logSessionID string, streamKey string) string {
+	return fmt.Sprintf("run.%s.%s.%s.%s", runEndpointID, serverInstanceID, logSessionID, streamKey)
+}
+
 func (svc *CoreService) GetJob(id string) (domain.Job, error) {
 	job, err := svc.store.Jobs().Get(id)
 	if err != nil {
@@ -2587,6 +2599,22 @@ func (svc *CoreService) ListArtifacts(filter domain.ArtifactFilter) ([]domain.Ar
 }
 
 func (svc *CoreService) CreateLogStream(stream domain.LogStream) (domain.LogStream, error) {
+	lock := svc.logIngestLock(stream.ServerInstanceID)
+	lock.Lock()
+	defer lock.Unlock()
+	return svc.createLogStream(stream)
+}
+
+func (svc *CoreService) logIngestLock(serverInstanceID string) *sync.Mutex {
+	hash := uint32(2166136261)
+	for index := 0; index < len(serverInstanceID); index++ {
+		hash ^= uint32(serverInstanceID[index])
+		hash *= 16777619
+	}
+	return &svc.logIngestMu[hash%uint32(len(svc.logIngestMu))]
+}
+
+func (svc *CoreService) createLogStream(stream domain.LogStream) (domain.LogStream, error) {
 	instance, err := svc.store.ServerInstances().Get(stream.ServerInstanceID)
 	if err != nil {
 		return domain.LogStream{}, fmt.Errorf("get server instance dependency: %w", err)
@@ -2604,12 +2632,31 @@ func (svc *CoreService) CreateLogStream(stream domain.LogStream) (domain.LogStre
 	if err := validator.ValidateLogStream(stream); err != nil {
 		return domain.LogStream{}, err
 	}
+	if err := svc.validateLogStreamSession(stream); err != nil {
+		return domain.LogStream{}, err
+	}
 	if err := svc.store.LogStreams().Create(stream); err != nil {
 		return domain.LogStream{}, err
 	}
 	return domain.CopyLogStream(stream), nil
 }
 
+func (svc *CoreService) validateLogStreamSession(stream domain.LogStream) error {
+	if stream.LogSessionID == "" {
+		return nil
+	}
+	streams, err := svc.store.LogStreams().List(domain.LogStreamFilter{ServerInstanceID: stream.ServerInstanceID})
+	if err != nil {
+		return err
+	}
+	for _, existing := range streams {
+		if existing.LogSessionID == stream.LogSessionID && !existing.SessionStartedAt.Equal(stream.SessionStartedAt) {
+			return validationError("log session metadata conflicts with an existing stream")
+		}
+	}
+	return nil
+}
+
 func (svc *CoreService) GetLogStream(id string) (domain.LogStream, error) {
 	return svc.store.LogStreams().Get(id)
 }