<commit_before>// Copyright 2017 Google 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 apiGatewayConfDeploy

import (
	"database/sql"
	"sync"

	"github.com/30x/apid-core"
)

var (
	gwBlobId int64
)

type DataDeployment struct {
	ID             string
	OrgID          string
	EnvID          string
	BlobID         string
	BlobResourceID string
	Type           string
	Name           string
	Revision       string
	Path           string
	Created        string
	CreatedBy      string
	Updated        string
	UpdatedBy      string
}

type SQLExec interface {
	Exec(query string, args ...interface{}) (sql.Result, error)
}

type dbManagerInterface interface {
	setDbVersion(string)
	initDb() error
	getUnreadyBlobs() ([]string, error)
	getReadyDeployments() ([]DataDeployment, error)
	updateLocalFsLocation(string, string) error
	getLocalFSLocation(string) (string, error)
}

type dbManager struct {
	data  apid.DataService
	db    apid.DB
	dbMux sync.RWMutex
}

func (dbc *dbManager) setDbVersion(version string) {
	db, err := dbc.data.DBVersion(version)
	if err != nil {
		log.Panicf("Unable to access database: %v", err)
	}
	dbc.dbMux.Lock()
	dbc.db = db
	dbc.dbMux.Unlock()
}

func (dbc *dbManager) getDb() apid.DB {
	dbc.dbMux.RLock()
	defer dbc.dbMux.RUnlock()
	return dbc.db
}

func (dbc *dbManager) initDb() error {
	_, err := dbc.getDb().Exec(`
	CREATE TABLE IF NOT EXISTS apid_blob_available (
		id text primary key,
   		local_fs_location text NOT NULL
	);
	`)
	if err != nil {
		return err
	}

	log.Debug("Database table apid_blob_available created.")
	return nil
}

// getUnreadyDeployments() returns array of resources that are not yet to be processed
// TODO make it work with new schema
func (dbc *dbManager) getUnreadyBlobs() (ids []string, err error) {

	// get unready blob id
	rows, err := dbc.getDb().Query(`
	SELECT a.bean_blob_id
		FROM metadata_runtime_entity_metadata as a
		LEFT JOIN apid_blob_available as b
		ON a.bean_blob_id = b.id
		WHERE b.id IS NULL;
	`)
	if err != nil {
		log.Errorf("DB Query for project_runtime_blob_metadata failed %v", err)
		return
	}
	defer rows.Close()
	for rows.Next() {
		var id string
		rows.Scan(&id)
		ids = append(ids, id)
	}

	// get unready resource id
	rows, err = dbc.getDb().Query(`
	SELECT a.resource_blob_id
		FROM metadata_runtime_entity_metadata as a
		LEFT JOIN apid_blob_available as b
		ON a.resource_blob_id = b.id
		WHERE (b.id IS NULL AND a.resource_blob_id IS NOT NULL AND a.resource_blob_id != '');
	`)
	if err != nil {
		log.Errorf("DB Query for project_runtime_blob_metadata failed %v", err)
		return
	}
	defer rows.Close()
	for rows.Next() {
		var id string
		rows.Scan(&id)
		ids = append(ids, id)
	}

	log.Debugf("Unready blobId %v", ids)
	return
}

// TODO there's a bug in the db statement
func (dbc *dbManager) getReadyDeployments() ([]DataDeployment, error) {


	rows, err := dbc.getDb().Query(`
		SELECT 	a.id,
			a.organization_id,
			a.environment_id,
			a.bean_blob_id,
			a.resource_blob_id,
			a.type,
			a.name,
			a.revision,
			a.path,
			a.created_at,
			a.created_by,
			a.updated_at,
			a.updated_by,
			b.local_fs_location
		FROM metadata_runtime_entity_metadata as a
		INNER JOIN edgex_blob_available as b
		ON (a.bean_blob_id = b.id)
		WHERE a.id IN (
			SELECT
					a.id
				FROM metadata_runtime_entity_metadata as a
				INNER JOIN edgex_blob_available as b
				ON a.resource_blob_id = b.id
				WHERE a.resource_blob_id != ""
			INTERSECT
				SELECT
					a.id
				FROM metadata_runtime_entity_metadata as a
				INNER JOIN edgex_blob_available as b
				ON a.resource_blob_id = b.id
				WHERE a.resource_blob_id != ""

			UNION
				SELECT
					a.id
				FROM metadata_runtime_entity_metadata as a
				INNER JOIN edgex_blob_available as b
				ON a.bean_blob_id = b.id
				WHERE a.resource_blob_id = ""
		)
	;
	`)

	if err != nil {
		log.Errorf("DB Query for project_runtime_blob_metadata failed %v", err)
		return nil, err
	}
	defer rows.Close()

	deployments, err := dataDeploymentsFromRow(rows)
	if err != nil {
		return nil, err
	}

	log.Debugf("Configurations ready: %v", deployments)

	if len(deployments) == 0 {
		log.Debug("No resources ready to be deployed")
		err = sql.ErrNoRows
	}<commit_msg>fix bug in db statement<commit_after>// Copyright 2017 Google 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 apiGatewayConfDeploy

import (
	"database/sql"
	"sync"

	"github.com/30x/apid-core"
)

var (
	gwBlobId int64
)

type DataDeployment struct {
	ID             string
	OrgID          string
	EnvID          string
	BlobID         string
	BlobResourceID string
	Type           string
	Name           string
	Revision       string
	Path           string
	Created        string
	CreatedBy      string
	Updated        string
	UpdatedBy      string
}

type SQLExec interface {
	Exec(query string, args ...interface{}) (sql.Result, error)
}

type dbManagerInterface interface {
	setDbVersion(string)
	initDb() error
	getUnreadyBlobs() ([]string, error)
	getReadyDeployments() ([]DataDeployment, error)
	updateLocalFsLocation(string, string) error
	getLocalFSLocation(string) (string, error)
}

type dbManager struct {
	data  apid.DataService
	db    apid.DB
	dbMux sync.RWMutex
}

func (dbc *dbManager) setDbVersion(version string) {
	db, err := dbc.data.DBVersion(version)
	if err != nil {
		log.Panicf("Unable to access database: %v", err)
	}
	dbc.dbMux.Lock()
	dbc.db = db
	dbc.dbMux.Unlock()
}

func (dbc *dbManager) getDb() apid.DB {
	dbc.dbMux.RLock()
	defer dbc.dbMux.RUnlock()
	return dbc.db
}

func (dbc *dbManager) initDb() error {
	_, err := dbc.getDb().Exec(`
	CREATE TABLE IF NOT EXISTS apid_blob_available (
		id text primary key,
   		local_fs_location text NOT NULL
	);
	`)
	if err != nil {
		return err
	}

	log.Debug("Database table apid_blob_available created.")
	return nil
}

// getUnreadyDeployments() returns array of resources that are not yet to be processed
// TODO make it work with new schema
func (dbc *dbManager) getUnreadyBlobs() (ids []string, err error) {

	// get unready blob id
	rows, err := dbc.getDb().Query(`
	SELECT a.bean_blob_id
		FROM metadata_runtime_entity_metadata as a
		LEFT JOIN apid_blob_available as b
		ON a.bean_blob_id = b.id
		WHERE b.id IS NULL;
	`)
	if err != nil {
		log.Errorf("DB Query for project_runtime_blob_metadata failed %v", err)
		return
	}
	defer rows.Close()
	for rows.Next() {
		var id string
		rows.Scan(&id)
		ids = append(ids, id)
	}

	// get unready resource id
	rows, err = dbc.getDb().Query(`
	SELECT a.resource_blob_id
		FROM metadata_runtime_entity_metadata as a
		LEFT JOIN apid_blob_available as b
		ON a.resource_blob_id = b.id
		WHERE (b.id IS NULL AND a.resource_blob_id IS NOT NULL AND a.resource_blob_id != '');
	`)
	if err != nil {
		log.Errorf("DB Query for project_runtime_blob_metadata failed %v", err)
		return
	}
	defer rows.Close()
	for rows.Next() {
		var id string
		rows.Scan(&id)
		ids = append(ids, id)
	}

	log.Debugf("Unready blobId %v", ids)
	return
}

func (dbc *dbManager) getReadyDeployments() ([]DataDeployment, error) {

	rows, err := dbc.getDb().Query(`
		SELECT 	a.id,
			a.organization_id,
			a.environment_id,
			a.bean_blob_id,
			a.resource_blob_id,
			a.type,
			a.name,
			a.revision,
			a.path,
			a.created_at,
			a.created_by,
			a.updated_at,
			a.updated_by
		FROM metadata_runtime_entity_metadata as a
		WHERE a.id IN (
			SELECT
					a.id
				FROM metadata_runtime_entity_metadata as a
				INNER JOIN apid_blob_available as b
				ON a.resource_blob_id = b.id
				WHERE a.resource_blob_id != ""
			INTERSECT
				SELECT
					a.id
				FROM metadata_runtime_entity_metadata as a
				INNER JOIN apid_blob_available as b
				ON a.bean_blob_id = b.id
				WHERE a.resource_blob_id != ""

			UNION
				SELECT
					a.id
				FROM metadata_runtime_entity_metadata as a
				INNER JOIN apid_blob_available as b
				ON a.bean_blob_id = b.id
				WHERE a.resource_blob_id = ""
		)
	;
	`)

	if err != nil {
		log.Errorf("DB Query for project_runtime_blob_metadata failed %v", err)
		return nil, err
	}
	defer rows.Close()

	deployments, err := dataDeploymentsFromRow(rows)
	if err != nil {
		return nil, err
	}

	log.Debugf("Configurations ready: %v", deployments)

	if len(deployments) == 0 {
		log.Debug("No resources ready to be deployed")
		err = sql.ErrNoRows
	}
{'fraction_non_alphanumeric': 0.09766044215496887, 'fraction_numerical': 0.002790298347284825, 'mean_word_length': 3.2186509732910817, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '4676594', 'n_tokens_mistral': 3486, 'n_tokens_neox': 3177, 'n_words': 965}