Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions stovepipe/core/filter/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -0,0 +1,9 @@
load("@rules_go//go:def.bzl", "go_library")

go_library(
name = "filter",
srcs = ["filter.go"],
importpath = "github.com/uber/submitqueue/stovepipe/core/filter",
visibility = ["//visibility:public"],
deps = ["//stovepipe/entity"],
)
23 changes: 21 additions & 2 deletions stovepipe/entity/entity.go → stovepipe/core/filter/filter.go
Original file line number Diff line number Diff line change
Expand Up @@ -12,5 +12,24 @@
// See the License for the specific language governing permissions and
// limitations under the License.

// Package entity holds Stovepipe-specific domain types (distinct from shared repo entity/).
package entity
// Package filter implements a filter for commit events.
package filter

import (
"strings"

"github.com/uber/submitqueue/stovepipe/entity"
)

// Config controls which VCS URIs are watched.
// WatchedURIPrefixes is a list of URI prefixes to match against ChangeEvent.URI.
// Example: "git://github.com/uber/go-code/refs/heads/main"
// watches all commits on the main branch of uber/go-code.
func ShouldProcess(event entity.ChangeEvent, watchedPrefixes []string) bool {
for _, prefix := range watchedPrefixes {
if strings.HasPrefix(event.URI, prefix) {
return true
}
}
return false
}
7 changes: 6 additions & 1 deletion stovepipe/entity/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,12 @@ load("@rules_go//go:def.bzl", "go_library")

go_library(
name = "entity",
srcs = ["entity.go"],
srcs = [
"change_event.go",
"change_uri.go",
"commit.go",
],
importpath = "github.com/uber/submitqueue/stovepipe/entity",
visibility = ["//visibility:public"],
deps = ["//stovepipe/entity/git"],
)
59 changes: 59 additions & 0 deletions stovepipe/entity/change_event.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,59 @@
// Copyright (c) 2025 Uber Technologies, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package entity

import (
"encoding/json"
"fmt"

entitygit "github.com/uber/submitqueue/stovepipe/entity/git"
)

// ChangeEvent represents a single new trunk change entering the pipeline, published to the
// start topic. A trunk change is one commit, so the event carries one git-backed URI. It is
// source-agnostic: both the webhook and the reconciliation poller emit it. Additional fields
// (e.g. source, committer time) can be added later as ingestion needs them.
type ChangeEvent struct {
// URI identifies the commit that entered the pipeline (git://owner/repo/branch/revision).
URI string `json:"uri"`
}

// ToBytes serializes the ChangeEvent to JSON bytes for queue message payload.
func (e ChangeEvent) ToBytes() ([]byte, error) {
return json.Marshal(e)
}

// Validate checks that the change event carries a valid git-backed commit URI.
func (e ChangeEvent) Validate() error {
if e.URI == "" {
return fmt.Errorf("change event requires a commit URI")
}
if _, err := entitygit.ParseChangeID(e.URI); err != nil {
return fmt.Errorf("change event URI: %w", err)
}
return nil
}

// ChangeEventFromBytes deserializes a ChangeEvent from JSON bytes.
func ChangeEventFromBytes(data []byte) (ChangeEvent, error) {
var event ChangeEvent
if err := json.Unmarshal(data, &event); err != nil {
return ChangeEvent{}, err
}
if err := event.Validate(); err != nil {
return ChangeEvent{}, err
}
return event, nil
}
36 changes: 36 additions & 0 deletions stovepipe/entity/change_uri.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,36 @@
// Copyright (c) 2025 Uber Technologies, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package entity

import "encoding/json"

// ChangeURI is the lightweight reference passed between pipeline stages. It
// carries only the change identity; stages re-resolve any state they need.
type ChangeURI struct {
// URI is the change identity (git://owner/repo/branch/revision).
URI string `json:"uri"`
}

// ToBytes serializes the ChangeURI to JSON bytes for a queue message payload.
func (c ChangeURI) ToBytes() ([]byte, error) {
return json.Marshal(c)
}

// ChangeURIFromBytes deserializes a ChangeURI from JSON bytes.
func ChangeURIFromBytes(data []byte) (ChangeURI, error) {
var ref ChangeURI
err := json.Unmarshal(data, &ref)
return ref, err
}
45 changes: 45 additions & 0 deletions stovepipe/entity/commit.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
// Copyright (c) 2025 Uber Technologies, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package entity

// CommitStatus is the validation state of a trunk commit as determined by Stovepipe.
type CommitStatus string

const (
// CommitStatusUnknown is the default state when a commit is first ingested.
// The commit has landed on main but has not yet been validated.
CommitStatusUnknown CommitStatus = ""
// CommitStatusSucceeded means the relevant targets build and test successfully at this commit.
CommitStatusSucceeded CommitStatus = "succeeded"
// CommitStatusFailed means a target is broken at this commit; it is the offending change.
CommitStatusFailed CommitStatus = "failed"
)

// IsCommitStatusTerminal returns true if the status is a final, irreversible state.
func IsCommitStatusTerminal(s CommitStatus) bool {
return s == CommitStatusSucceeded || s == CommitStatusFailed
}

// Commit is a trunk commit tracked by Stovepipe's gateway.
type Commit struct {
// URI is the canonical change identity from the originating ChangeEvent.
URI string
// Status is the current validation state of this commit.
Status CommitStatus
// CreatedAt is the time this commit was first recorded, in milliseconds since epoch.
CreatedAt int64
// UpdatedAt is the time this commit was last updated, in milliseconds since epoch.
UpdatedAt int64
}
1 change: 1 addition & 0 deletions stovepipe/extension/BUILD.bazel
Original file line number Diff line number Diff line change
Expand Up @@ -5,4 +5,5 @@ go_library(
srcs = ["extension.go"],
importpath = "github.com/uber/submitqueue/stovepipe/extension",
visibility = ["//visibility:public"],
deps = ["//stovepipe/entity"],
)
13 changes: 13 additions & 0 deletions stovepipe/extension/changeingester/BUILD.bazel
Original file line number Diff line number Diff line change
@@ -0,0 +1,13 @@
load("@rules_go//go:def.bzl", "go_library")

go_library(
name = "changeingester",
srcs = ["logging.go"],
importpath = "github.com/uber/submitqueue/stovepipe/extension/changeingester",
visibility = ["//visibility:public"],
deps = [
"//stovepipe/entity",
"//stovepipe/extension",
"@org_uber_go_zap//:zap",
],
)
43 changes: 43 additions & 0 deletions stovepipe/extension/changeingester/logging.go
Original file line number Diff line number Diff line change
@@ -0,0 +1,43 @@
// Copyright (c) 2025 Uber Technologies, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

// Package changeingester provides ChangeIngester implementations.
package changeingester

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

i think before we go to the extensions, we need to think from queue perspective,

stream of changes ---> some queue (change-ingester) ---> change-ingester-controller ---> aggregate -> create Batch entity ---> speculate --> push or fix/revert

@mnoah1 maybe it would we worth starting from a workflow doc (similar to what we have submitqueue) before we jump into the code... that would lay out the end to end flow of various components to make the code changes easier.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@behinddwalls @mnoah1 let me know if I can help with this


import (
"context"

"github.com/uber/submitqueue/stovepipe/entity"
"github.com/uber/submitqueue/stovepipe/extension"
"go.uber.org/zap"
)

// LoggingHandler is a stub ChangeHandler that logs received changes.
// Replace with real persistence logic once DB schema is ready.
type LoggingHandler struct {
logger *zap.Logger
}

// New constructs a new LoggingHandler.
// The return type enforces interface compliance at compile time.
func New(logger *zap.Logger) extension.ChangeHandler {
return LoggingHandler{logger: logger}
}

func (h LoggingHandler) IngestChange(ctx context.Context, event entity.ChangeEvent) error {
h.logger.Info("ingested change",
zap.String("uri", event.URI),
)
return nil
}
18 changes: 18 additions & 0 deletions stovepipe/extension/extension.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,3 +14,21 @@

// Package extension holds Stovepipe-specific extension implementations.
package extension

import (
"context"

"github.com/uber/submitqueue/stovepipe/entity"
)

// ChangeIngester subscribes to change events from a VCS source
// and dispatches them for processing. The source and VCS are
// implementation details left to the injected backend.
type ChangeIngester interface {
Start(ctx context.Context) error
}

// ChangeHandler processes a single change received from the ingester.
type ChangeHandler interface {
IngestChange(ctx context.Context, event entity.ChangeEvent) error
}
Loading