<commit_before>package db

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	"code.cloudfoundry.org/lager"

	sq "github.com/Masterminds/squirrel"
	"github.com/concourse/atc"
	"github.com/concourse/atc/db/algorithm"
	"github.com/concourse/atc/db/lock"
)

type ErrResourceNotFound struct {
	Name string
}

func (e ErrResourceNotFound) Error() string {
	return fmt.Sprintf("resource '%s' not found", e.Name)
}

//go:generate counterfeiter . Pipeline

type Cause struct {
	VersionedResourceID int `json:"versioned_resource_id"`
	BuildID             int `json:"build_id"`
}

type Pipeline interface {
	ID() int
	Name() string
	TeamID() int
	TeamName() string
	Groups() atc.GroupConfigs
	ConfigVersion() ConfigVersion
	Public() bool
	Paused() bool
	ScopedName(string) string

	CheckPaused() (bool, error)
	Reload() (bool, error)

	Causality(versionedResourceID int) ([]Cause, error)

	SetResourceCheckError(Resource, error) error
	SaveResourceVersions(atc.ResourceConfig, []atc.Version) error
	GetResourceVersions(resourceName string, page Page) ([]SavedVersionedResource, Pagination, bool, error)

	GetAllPendingBuilds() (map[string][]Build, error)

	GetLatestVersionedResource(resourceName string) (SavedVersionedResource, bool, error)
	GetVersionedResourceByVersion(atcVersion atc.Version, resourceName string) (SavedVersionedResource, bool, error)

	VersionedResource(versionedResourceID int) (SavedVersionedResource, bool, error)
	DisableVersionedResource(versionedResourceID int) error
	EnableVersionedResource(versionedResourceID int) error
	GetBuildsWithVersionAsInput(versionedResourceID int) ([]Build, error)
	GetBuildsWithVersionAsOutput(versionedResourceID int) ([]Build, error)

	DeleteBuildEventsByBuildIDs(buildIDs []int) error

	// Needs test (from db/lock_test.go)
	AcquireSchedulingLock(lager.Logger, time.Duration) (lock.Lock, bool, error)

	AcquireResourceCheckingLockWithIntervalCheck(
		logger lager.Logger,
		resourceName string,
		usedResourceConfig *UsedResourceConfig,
		interval time.Duration,
		immediate bool,
	) (lock.Lock, bool, error)

	AcquireResourceTypeCheckingLockWithIntervalCheck(
		logger lager.Logger,
		resourceTypeName string,
		usedResourceConfig *UsedResourceConfig,
		interval time.Duration,
		immediate bool,
	) (lock.Lock, bool, error)

	LoadVersionsDB() (*algorithm.VersionsDB, error)

	Resource(name string) (Resource, bool, error)
	Resources() (Resources, error)

	ResourceTypes() (ResourceTypes, error)
	ResourceType(name string) (ResourceType, bool, error)

	Job(name string) (Job, bool, error)
	Jobs() (Jobs, error)
	Dashboard() (Dashboard, atc.GroupConfigs, error)

	Expose() error
	Hide() error

	Pause() error
	Unpause() error

	Destroy() error
	Rename(string) error

	CreateOneOffBuild() (Build, error)
}

type pipeline struct {
	id            int
	name          string
	teamID        int
	teamName      string
	groups        atc.GroupConfigs
	configVersion ConfigVersion
	paused        bool
	public        bool

	cachedAt   time.Time
	versionsDB *algorithm.VersionsDB

	conn        Conn
	lockFactory lock.LockFactory
}

//ConfigVersion is a sequence identifier used for compare-and-swap
type ConfigVersion int

type PipelinePausedState string

var pipelinesQuery = psql.Select(`
		p.id,
		p.name,
		p.groups,
		p.version,
		p.team_id,
		t.name,
		p.paused,
		p.public
	`).
	From("pipelines p").
	LeftJoin("teams t ON p.team_id = t.id")

const (
	PipelinePaused   PipelinePausedState = "paused"
	PipelineUnpaused PipelinePausedState = "unpaused"
	PipelineNoChange PipelinePausedState = "nochange"
)

func (state PipelinePausedState) Bool() *bool {
	yes := true
	no := false

	switch state {
	case PipelinePaused:
		return &yes
	case PipelineUnpaused:
		return &no
	case PipelineNoChange:
		return nil
	default:
		panic("unknown pipeline state")
	}
}

func newPipeline(conn Conn, lockFactory lock.LockFactory) *pipeline {
	return &pipeline{
		conn:        conn,
		lockFactory: lockFactory,
	}
}

func (p *pipeline) ID() int                      { return p.id }
func (p *pipeline) Name() string                 { return p.name }
func (p *pipeline) TeamID() int                  { return p.teamID }
func (p *pipeline) TeamName() string             { return p.teamName }
func (p *pipeline) Groups() atc.GroupConfigs     { return p.groups }
func (p *pipeline) ConfigVersion() ConfigVersion { return p.configVersion }
func (p *pipeline) Public() bool                 { return p.public }
func (p *pipeline) Paused() bool                 { return p.paused }

func (p *pipeline) ScopedName(n string) string {
	return p.name + ":" + n
}

func (p *pipeline) Causality(versionedResourceID int) ([]Cause, error) {
	rows, err := p.conn.Query(`
		WITH RECURSIVE causality(versioned_resource_id, build_id) AS (
				SELECT bi.versioned_resource_id, bi.build_id
				FROM build_inputs bi
				WHERE bi.versioned_resource_id = $1
			UNION
				SELECT bi.versioned_resource_id, bi.build_id
				FROM build_inputs bi, build_outputs bo, causality c
				WHERE bi.versioned_resource_id = bo.versioned_resource_id
				AND bo.build_id = c.build_id
				AND bo.explicit
		)
		SELECT versioned_resource_id, build_id
		FROM causality
		ORDER BY build_id ASC, versioned_resource_id ASC
	`, versionedResourceID)
	if err != nil {
		return nil, err
	}

	var causality []Cause
	for rows.Next() {<commit_msg>fix ever-growing chain of causality from circles<commit_after>package db

import (
	"database/sql"
	"encoding/json"
	"fmt"
	"strconv"
	"strings"
	"time"

	"code.cloudfoundry.org/lager"

	sq "github.com/Masterminds/squirrel"
	"github.com/concourse/atc"
	"github.com/concourse/atc/db/algorithm"
	"github.com/concourse/atc/db/lock"
)

type ErrResourceNotFound struct {
	Name string
}

func (e ErrResourceNotFound) Error() string {
	return fmt.Sprintf("resource '%s' not found", e.Name)
}

//go:generate counterfeiter . Pipeline

type Cause struct {
	VersionedResourceID int `json:"versioned_resource_id"`
	BuildID             int `json:"build_id"`
}

type Pipeline interface {
	ID() int
	Name() string
	TeamID() int
	TeamName() string
	Groups() atc.GroupConfigs
	ConfigVersion() ConfigVersion
	Public() bool
	Paused() bool
	ScopedName(string) string

	CheckPaused() (bool, error)
	Reload() (bool, error)

	Causality(versionedResourceID int) ([]Cause, error)

	SetResourceCheckError(Resource, error) error
	SaveResourceVersions(atc.ResourceConfig, []atc.Version) error
	GetResourceVersions(resourceName string, page Page) ([]SavedVersionedResource, Pagination, bool, error)

	GetAllPendingBuilds() (map[string][]Build, error)

	GetLatestVersionedResource(resourceName string) (SavedVersionedResource, bool, error)
	GetVersionedResourceByVersion(atcVersion atc.Version, resourceName string) (SavedVersionedResource, bool, error)

	VersionedResource(versionedResourceID int) (SavedVersionedResource, bool, error)
	DisableVersionedResource(versionedResourceID int) error
	EnableVersionedResource(versionedResourceID int) error
	GetBuildsWithVersionAsInput(versionedResourceID int) ([]Build, error)
	GetBuildsWithVersionAsOutput(versionedResourceID int) ([]Build, error)

	DeleteBuildEventsByBuildIDs(buildIDs []int) error

	// Needs test (from db/lock_test.go)
	AcquireSchedulingLock(lager.Logger, time.Duration) (lock.Lock, bool, error)

	AcquireResourceCheckingLockWithIntervalCheck(
		logger lager.Logger,
		resourceName string,
		usedResourceConfig *UsedResourceConfig,
		interval time.Duration,
		immediate bool,
	) (lock.Lock, bool, error)

	AcquireResourceTypeCheckingLockWithIntervalCheck(
		logger lager.Logger,
		resourceTypeName string,
		usedResourceConfig *UsedResourceConfig,
		interval time.Duration,
		immediate bool,
	) (lock.Lock, bool, error)

	LoadVersionsDB() (*algorithm.VersionsDB, error)

	Resource(name string) (Resource, bool, error)
	Resources() (Resources, error)

	ResourceTypes() (ResourceTypes, error)
	ResourceType(name string) (ResourceType, bool, error)

	Job(name string) (Job, bool, error)
	Jobs() (Jobs, error)
	Dashboard() (Dashboard, atc.GroupConfigs, error)

	Expose() error
	Hide() error

	Pause() error
	Unpause() error

	Destroy() error
	Rename(string) error

	CreateOneOffBuild() (Build, error)
}

type pipeline struct {
	id            int
	name          string
	teamID        int
	teamName      string
	groups        atc.GroupConfigs
	configVersion ConfigVersion
	paused        bool
	public        bool

	cachedAt   time.Time
	versionsDB *algorithm.VersionsDB

	conn        Conn
	lockFactory lock.LockFactory
}

//ConfigVersion is a sequence identifier used for compare-and-swap
type ConfigVersion int

type PipelinePausedState string

var pipelinesQuery = psql.Select(`
		p.id,
		p.name,
		p.groups,
		p.version,
		p.team_id,
		t.name,
		p.paused,
		p.public
	`).
	From("pipelines p").
	LeftJoin("teams t ON p.team_id = t.id")

const (
	PipelinePaused   PipelinePausedState = "paused"
	PipelineUnpaused PipelinePausedState = "unpaused"
	PipelineNoChange PipelinePausedState = "nochange"
)

func (state PipelinePausedState) Bool() *bool {
	yes := true
	no := false

	switch state {
	case PipelinePaused:
		return &yes
	case PipelineUnpaused:
		return &no
	case PipelineNoChange:
		return nil
	default:
		panic("unknown pipeline state")
	}
}

func newPipeline(conn Conn, lockFactory lock.LockFactory) *pipeline {
	return &pipeline{
		conn:        conn,
		lockFactory: lockFactory,
	}
}

func (p *pipeline) ID() int                      { return p.id }
func (p *pipeline) Name() string                 { return p.name }
func (p *pipeline) TeamID() int                  { return p.teamID }
func (p *pipeline) TeamName() string             { return p.teamName }
func (p *pipeline) Groups() atc.GroupConfigs     { return p.groups }
func (p *pipeline) ConfigVersion() ConfigVersion { return p.configVersion }
func (p *pipeline) Public() bool                 { return p.public }
func (p *pipeline) Paused() bool                 { return p.paused }

func (p *pipeline) ScopedName(n string) string {
	return p.name + ":" + n
}

func (p *pipeline) Causality(versionedResourceID int) ([]Cause, error) {
	rows, err := p.conn.Query(`
		WITH RECURSIVE transitive_outputs(versioned_resource_id, build_id) AS (
				SELECT bi.versioned_resource_id, bi.build_id
				FROM build_inputs bi
				WHERE bi.versioned_resource_id = $1
			UNION
				SELECT bi.versioned_resource_id, bi.build_id
				FROM build_inputs bi, build_outputs bo, transitive_outputs t
				WHERE bi.versioned_resource_id = bo.versioned_resource_id
				AND bo.build_id = t.build_id
				AND bo.explicit
		), first_version_of_each_resource AS (
			SELECT min(versioned_resource_id) AS versioned_resource_id
			FROM transitive_outputs t
			LEFT JOIN versioned_resources v ON v.id = t.versioned_resource_id
			GROUP BY v.resource_id
		), truncated_to_earliest_path AS (
			SELECT t.versioned_resource_id, t.build_id
			FROM transitive_outputs t, first_version_of_each_resource f
			WHERE t.versioned_resource_id = f.versioned_resource_id
		)
		SELECT versioned_resource_id, build_id
		FROM truncated_to_earliest_path
		ORDER BY build_id ASC, versioned_resource_id ASC
	`, versionedResourceID)
	if err != nil {
		return nil, err
	}

	var causality []Cause
	for rows.Next() {
{'fraction_non_alphanumeric': 0.09623281556864846, 'fraction_numerical': 0.00017853954650955185, 'mean_word_length': 4.304450757575758, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4282778', 'n_tokens_mistral': 3864, 'n_tokens_neox': 3638, 'n_words': 932}