more work on worker integration

server/datastore/migrate/helper.go

@@ -0,0 +1,46 @@
+package migrate
+
+import (
+	"strconv"
+	"strings"
+
+	"github.com/russross/meddler"
+)
+
+// transform is a helper function that transforms sql
+// statements to work with multiple database types.
+func transform(stmt string) string {
+	switch meddler.Default {
+	case meddler.MySQL:
+		stmt = strings.Replace(stmt, "AUTOINCREMENT", "AUTO_INCREMENT", -1)
+	case meddler.PostgreSQL:
+		stmt = strings.Replace(stmt, "INTEGER PRIMARY KEY AUTOINCREMENT", "SERIAL PRIMARY KEY", -1)
+		stmt = strings.Replace(stmt, "BLOB", "BYTEA", -1)
+	}
+	return stmt
+}
+
+// rebind is a helper function that changes the sql
+// bind type from ? to $ for postgres queries.
+func rebind(query string) string {
+	if meddler.Default != meddler.PostgreSQL {
+		return query
+	}
+
+	qb := []byte(query)
+	// Add space enough for 10 params before we have to allocate
+	rqb := make([]byte, 0, len(qb)+10)
+	j := 1
+	for _, b := range qb {
+		if b == '?' {
+			rqb = append(rqb, '$')
+			for _, b := range strconv.Itoa(j) {
+				rqb = append(rqb, byte(b))
+			}
+			j++
+		} else {
+			rqb = append(rqb, b)
+		}
+	}
+	return string(rqb)
+}

server/datastore/migrate/setup.go

@@ -0,0 +1,118 @@
+package migrate
+
+import (
+	"github.com/BurntSushi/migration"
+)
+
+// Setup is the database migration function that
+// will setup the initial SQL database structure.
+func Setup(tx migration.LimitedTx) error {
+	var stmts = []string{
+		blobTable,
+		userTable,
+		repoTable,
+		permTable,
+		commitTable,
+	}
+	for _, stmt := range stmts {
+		_, err := tx.Exec(transform(stmt))
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+var userTable = `
+CREATE TABLE IF NOT EXISTS users (
+	 user_id           INTEGER PRIMARY KEY AUTOINCREMENT
+	,user_remote       VARCHAR(255)
+	,user_login        VARCHAR(255)
+	,user_access       VARCHAR(255)
+	,user_secret       VARCHAR(255)
+	,user_name         VARCHAR(255)
+	,user_email        VARCHAR(255)
+	,user_gravatar     VARCHAR(255)
+	,user_token        VARCHAR(255)
+	,user_admin        BOOLEAN
+	,user_active       BOOLEAN
+	,user_syncing      BOOLEAN
+	,user_created      INTEGER
+	,user_updated      INTEGER
+	,user_synced       INTEGER
+	,UNIQUE(user_token)
+	,UNIQUE(user_remote, user_login)
+);
+`
+
+var permTable = `
+CREATE TABLE IF NOT EXISTS perms (
+	 perm_id           INTEGER PRIMARY KEY AUTOINCREMENT
+	,user_id           INTEGER
+	,repo_id           INTEGER
+	,perm_read         BOOLEAN
+	,perm_write        BOOLEAN
+	,perm_admin        BOOLEAN
+	,perm_created      INTEGER
+	,perm_updated      INTEGER
+	,UNIQUE (repo_id, user_id)
+);
+`
+
+var repoTable = `
+CREATE TABLE IF NOT EXISTS repos (
+	 repo_id           INTEGER PRIMARY KEY AUTOINCREMENT
+	,user_id           INTEGER
+	,repo_remote       VARCHAR(255)
+	,repo_host         VARCHAR(255)
+	,repo_owner        VARCHAR(255)
+	,repo_name         VARCHAR(255)
+	,repo_url          VARCHAR(1024)
+	,repo_clone_url    VARCHAR(255)
+	,repo_git_url      VARCHAR(255)
+	,repo_ssh_url      VARCHAR(255)
+	,repo_active       BOOLEAN
+	,repo_private      BOOLEAN
+	,repo_privileged   BOOLEAN
+	,repo_post_commit  BOOLEAN
+	,repo_pull_request BOOLEAN
+	,repo_public_key   BLOB
+	,repo_private_key  BLOB
+	,repo_params       BLOB
+	,repo_timeout      INTEGER
+	,repo_created      INTEGER
+	,repo_updated      INTEGER
+	,UNIQUE(repo_host, repo_owner, repo_name)
+);
+`
+
+var commitTable = `
+CREATE TABLE IF NOT EXISTS commits (
+	 commit_id         INTEGER PRIMARY KEY AUTOINCREMENT
+	,repo_id           INTEGER
+	,commit_status     VARCHAR(255)
+	,commit_started    INTEGER
+	,commit_finished   INTEGER
+	,commit_duration   INTEGER
+	,commit_sha        VARCHAR(255)
+	,commit_branch     VARCHAR(255)
+	,commit_pr         VARCHAR(255)
+	,commit_author     VARCHAR(255)
+	,commit_gravatar   VARCHAR(255)
+	,commit_timestamp  VARCHAR(255)
+	,commit_message    VARCHAR(255)
+	,commit_yaml       BLOB
+	,commit_created    INTEGER
+	,commit_updated    INTEGER
+	,UNIQUE(commit_sha, commit_branch, repo_id)
+);
+`
+
+var blobTable = `
+CREATE TABLE IF NOT EXISTS blobs (
+	 blob_id      INTEGER PRIMARY KEY AUTOINCREMENT
+	,blob_path    VARCHAR(1024)
+	,blob_data    BLOB
+	,UNIQUE(blob_path)
+);
+`

server/datastore/migrate/version.go

@@ -0,0 +1,57 @@
+package migrate
+
+import (
+	"github.com/BurntSushi/migration"
+)
+
+// GetVersion gets the migration version from the database,
+// creating the migration table if it does not already exist.
+func GetVersion(tx migration.LimitedTx) (int, error) {
+	v, err := getVersion(tx)
+	if err != nil {
+		if err := createVersionTable(tx); err != nil {
+			return 0, err
+		}
+		return getVersion(tx)
+	}
+	return v, nil
+}
+
+// SetVersion sets the migration version in the database,
+// creating the migration table if it does not already exist.
+func SetVersion(tx migration.LimitedTx, version int) error {
+	if err := setVersion(tx, version); err != nil {
+		if err := createVersionTable(tx); err != nil {
+			return err
+		}
+		return setVersion(tx, version)
+	}
+	return nil
+}
+
+// setVersion updates the migration version in the database.
+func setVersion(tx migration.LimitedTx, version int) error {
+	_, err := tx.Exec(rebind("UPDATE migration_version SET version = ?"), version)
+	return err
+}
+
+// getVersion gets the migration version in the database.
+func getVersion(tx migration.LimitedTx) (int, error) {
+	var version int
+	row := tx.QueryRow("SELECT version FROM migration_version")
+	if err := row.Scan(&version); err != nil {
+		return 0, err
+	}
+	return version, nil
+}
+
+// createVersionTable creates the version table and inserts the
+// initial value (0) into the database.
+func createVersionTable(tx migration.LimitedTx) error {
+	_, err := tx.Exec("CREATE TABLE migration_version ( version INTEGER )")
+	if err != nil {
+		return err
+	}
+	_, err = tx.Exec("INSERT INTO migration_version (version) VALUES (0)")
+	return err
+}

server/handler/commit.go

@@ -5,6 +5,9 @@ import (
 	"net/http"
 
 	"github.com/drone/drone/server/datastore"
+	"github.com/drone/drone/server/worker"
+	"github.com/drone/drone/shared/httputil"
+	"github.com/drone/drone/shared/model"
 	"github.com/goji/context"
 	"github.com/zenazn/goji/web"
 )
@@ -50,71 +53,55 @@ func GetCommit(c web.C, w http.ResponseWriter, r *http.Request) {
 	json.NewEncoder(w).Encode(commit)
 }
 
-func PostCommit(c web.C, w http.ResponseWriter, r *http.Request) {}
+// PostHook accepts a post-commit hook and parses the payload
+// in order to trigger a build. The payload is specified to the
+// remote system (ie GitHub) and will therefore get parsed by
+// the appropriate remote plugin.
+//
+//     POST /api/repos/{host}/{owner}/{name}/branches/{branch}/commits/{commit}
+//
+func PostCommit(c web.C, w http.ResponseWriter, r *http.Request) {
+	var ctx = context.FromC(c)
+	var (
+		branch = c.URLParams["branch"]
+		hash   = c.URLParams["commit"]
+		repo   = ToRepo(c)
+	)
 
-/*
-// PostCommit gets the commit for the repository and schedules to re-build.
-// GET /v1/repos/{host}/{owner}/{name}/branches/{branch}/commits/{commit}
-func (h *CommitHandler) PostCommit(w http.ResponseWriter, r *http.Request) error {
-	var host, owner, name = parseRepo(r)
-	var branch = r.FormValue(":branch")
-	var sha = r.FormValue(":commit")
-
-	// get the user form the session.
-	user := h.sess.User(r)
-	if user == nil {
-		return notAuthorized{}
-	}
-
-	// get the repo from the database
-	repo, err := h.repos.FindName(host, owner, name)
-	switch {
-	case err != nil && user == nil:
-		return notAuthorized{}
-	case err != nil && user != nil:
-		return notFound{}
-	}
-
-	// user must have admin access to the repository.
-	if ok, _ := h.perms.Admin(user, repo); !ok {
-		return notFound{err}
-	}
-
-	c, err := h.commits.FindSha(repo.ID, branch, sha)
+	commit, err := datastore.GetCommitSha(ctx, repo, branch, hash)
 	if err != nil {
-		return notFound{err}
+		w.WriteHeader(http.StatusNotFound)
+		return
 	}
 
-	// we can't start an already started build
-	if c.Status == model.StatusStarted || c.Status == model.StatusEnqueue {
-		return badRequest{}
+	if commit.Status == model.StatusStarted ||
+		commit.Status == model.StatusEnqueue {
+		w.WriteHeader(http.StatusConflict)
+		return
 	}
 
-	c.Status = model.StatusEnqueue
-	c.Started = 0
-	c.Finished = 0
-	c.Duration = 0
-	if err := h.commits.Update(c); err != nil {
-		return internalServerError{err}
+	commit.Status = model.StatusEnqueue
+	commit.Started = 0
+	commit.Finished = 0
+	commit.Duration = 0
+	if err := datastore.PutCommit(ctx, commit); err != nil {
+		w.WriteHeader(http.StatusInternalServerError)
+		return
 	}
 
-	repoOwner, err := h.users.Find(repo.UserID)
+	owner, err := datastore.GetUser(ctx, repo.UserID)
 	if err != nil {
-		return badRequest{err}
+		w.WriteHeader(http.StatusBadRequest)
+		return
 	}
 
 	// drop the items on the queue
-	// drop the items on the queue
-	go func() {
-		h.queue <- &model.Request{
-			User:   repoOwner,
-			Host:   httputil.GetURL(r),
+	go worker.Do(ctx, &worker.Work{
+		User:   owner,
 		Repo:   repo,
-			Commit: c,
-		}
-	}()
+		Commit: commit,
+		Host:   httputil.GetURL(r),
+	})
 
 	w.WriteHeader(http.StatusOK)
-	return nil
 }
-*/

server/handler/hook.go

@@ -1,3 +1,116 @@
 package handler
 
-// PostHook
+import (
+	"net/http"
+	"strings"
+
+	"github.com/drone/drone/plugin/remote"
+	"github.com/drone/drone/server/datastore"
+	"github.com/drone/drone/server/worker"
+	"github.com/drone/drone/shared/build/script"
+	"github.com/drone/drone/shared/httputil"
+	"github.com/drone/drone/shared/model"
+	"github.com/goji/context"
+	"github.com/zenazn/goji/web"
+)
+
+// PostHook accepts a post-commit hook and parses the payload
+// in order to trigger a build. The payload is specified to the
+// remote system (ie GitHub) and will therefore get parsed by
+// the appropriate remote plugin.
+//
+//     GET /api/hook/:host
+//
+func PostHook(c web.C, w http.ResponseWriter, r *http.Request) {
+	var ctx = context.FromC(c)
+	var host = c.URLParams["host"]
+	var remote = remote.Lookup(host)
+	if remote == nil {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	// parse the hook payload
+	hook, err := remote.ParseHook(r)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	// in some cases we have neither a hook nor error. An example
+	// would be GitHub sending a ping request to the URL, in which
+	// case we'll just exit quiely with an 'OK'
+	if hook == nil || strings.Contains(hook.Message, "[CI SKIP]") {
+		w.WriteHeader(http.StatusOK)
+		return
+	}
+
+	// fetch the repository from the database
+	repo, err := datastore.GetRepoName(ctx, remote.GetHost(), hook.Owner, hook.Repo)
+	if err != nil {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	if repo.Active == false ||
+		(repo.PostCommit == false && len(hook.PullRequest) == 0) ||
+		(repo.PullRequest == false && len(hook.PullRequest) != 0) {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	// fetch the user from the database that owns this repo
+	user, err := datastore.GetUser(ctx, repo.UserID)
+	if err != nil {
+		w.WriteHeader(http.StatusNotFound)
+		return
+	}
+
+	// featch the .drone.yml file from the database
+	yml, err := remote.GetScript(user, repo, hook)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	// verify the commit hooks branch matches the list of approved
+	// branches (unless it is a pull request). Note that we don't really
+	// care if parsing the yaml fails here.
+	s, _ := script.ParseBuild(string(yml), map[string]string{})
+	if len(hook.PullRequest) == 0 && !s.MatchBranch(hook.Branch) {
+		w.WriteHeader(http.StatusOK)
+		return
+	}
+
+	commit := model.Commit{
+		RepoID:      repo.ID,
+		Status:      model.StatusEnqueue,
+		Sha:         hook.Sha,
+		Branch:      hook.Branch,
+		PullRequest: hook.PullRequest,
+		Timestamp:   hook.Timestamp,
+		Message:     hook.Message,
+		Config:      string(yml)}
+	commit.SetAuthor(hook.Author)
+	// inser the commit into the database
+	if err := datastore.PostCommit(ctx, &commit); err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	owner, err := datastore.GetUser(ctx, repo.UserID)
+	if err != nil {
+		w.WriteHeader(http.StatusBadRequest)
+		return
+	}
+
+	// drop the items on the queue
+	go worker.Do(ctx, &worker.Work{
+		User:   owner,
+		Repo:   repo,
+		Commit: &commit,
+		Host:   httputil.GetURL(r),
+	})
+
+	w.WriteHeader(http.StatusOK)
+}

server/main.go

@@ -31,6 +31,9 @@ import (
 	"github.com/drone/drone/server/blobstore"
 	"github.com/drone/drone/server/datastore"
 	"github.com/drone/drone/server/datastore/database"
+	"github.com/drone/drone/server/worker/director"
+	"github.com/drone/drone/server/worker/docker"
+	"github.com/drone/drone/server/worker/pool"
 )
 
 var (
@@ -52,15 +55,17 @@ var (
 	version  string = "0.3-dev"
 	revision string
 
-	// Number of concurrent build workers to run
-	// default to number of CPUs on machine
-	workers int
-
 	conf   string
 	prefix string
 
 	open bool
 
+	// worker pool
+	workers *pool.Pool
+
+	// director
+	worker *director.Director
+
 	nodes StringArr
 
 	db *sql.DB
@@ -92,6 +97,26 @@ func main() {
 	db = database.MustConnect(driver, datasource)
 	go database.NewCommitstore(db).KillCommits()
 
+	// Create the worker, director and builders
+	workers = pool.New()
+	workers.Allocate(docker.New())
+	workers.Allocate(docker.New())
+	workers.Allocate(docker.New())
+	workers.Allocate(docker.New())
+	worker = director.New()
+
+	/*
+		if nodes == nil || len(nodes) == 0 {
+			worker.NewWorker(workerc, users, repos, commits, pubsub, &model.Server{}).Start()
+			worker.NewWorker(workerc, users, repos, commits, pubsub, &model.Server{}).Start()
+		} else {
+			for _, node := range nodes {
+				println(node)
+				worker.NewWorker(workerc, users, repos, commits, pubsub, &model.Server{Host: node}).Start()
+			}
+		}
+	*/
+
 	goji.Get("/api/auth/:host", handler.GetLogin)
 	goji.Get("/api/badge/:host/:owner/:name/status.svg", handler.GetBadge)
 	goji.Get("/api/badge/:host/:owner/:name/cc.xml", handler.GetCC)
@@ -135,20 +160,6 @@ func main() {
 	goji.Use(middleware.SetUser)
 	goji.Serve()
 
-	// if no worker nodes are specified than start 2 workers
-	// using the default DOCKER_HOST
-	/*
-		if nodes == nil || len(nodes) == 0 {
-			worker.NewWorker(workerc, users, repos, commits, pubsub, &model.Server{}).Start()
-			worker.NewWorker(workerc, users, repos, commits, pubsub, &model.Server{}).Start()
-		} else {
-			for _, node := range nodes {
-				println(node)
-				worker.NewWorker(workerc, users, repos, commits, pubsub, &model.Server{Host: node}).Start()
-			}
-		}
-	*/
-
 	// start webserver using HTTPS or HTTP
 	//if len(sslcert) != 0 {
 	//	panic(http.ListenAndServeTLS(port, sslcert, sslkey, nil))
@@ -164,8 +175,8 @@ func ContextMiddleware(c *web.C, h http.Handler) http.Handler {
 		var ctx = context.Background()
 		ctx = datastore.NewContext(ctx, database.NewDatastore(db))
 		ctx = blobstore.NewContext(ctx, database.NewBlobstore(db))
-		//ctx = pool.NewContext(ctx, workers)
-		//ctx = director.NewContext(ctx, worker)
+		ctx = pool.NewContext(ctx, workers)
+		ctx = director.NewContext(ctx, worker)
 
 		// add the context to the goji web context
 		webcontext.Set(c, ctx)

server/worker/work.go

@@ -3,6 +3,7 @@ package worker
 import "github.com/drone/drone/shared/model"
 
 type Work struct {
+	Host   string
 	User   *model.User
 	Repo   *model.Repo
 	Commit *model.Commit