sftpgo/internal/dataprovider/sqlcommon.go
Nicola Murino 784b7585c1
remove end year from Copyright notice in files
so we don't have to update all the files every year

Signed-off-by: Nicola Murino <nicola.murino@gmail.com>
2024-01-01 11:31:45 +01:00

4041 lines
110 KiB
Go

// Copyright (C) 2019 Nicola Murino
//
// This program is free software: you can redistribute it and/or modify
// it under the terms of the GNU Affero General Public License as published
// by the Free Software Foundation, version 3.
//
// This program is distributed in the hope that it will be useful,
// but WITHOUT ANY WARRANTY; without even the implied warranty of
// MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
// GNU Affero General Public License for more details.
//
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see <https://www.gnu.org/licenses/>.
package dataprovider
import (
"context"
"crypto/x509"
"database/sql"
"encoding/json"
"errors"
"fmt"
"net/netip"
"runtime/debug"
"strings"
"time"
"github.com/cockroachdb/cockroach-go/v2/crdb"
"github.com/sftpgo/sdk"
"github.com/drakkan/sftpgo/v2/internal/logger"
"github.com/drakkan/sftpgo/v2/internal/util"
"github.com/drakkan/sftpgo/v2/internal/vfs"
)
const (
sqlDatabaseVersion = 28
defaultSQLQueryTimeout = 10 * time.Second
longSQLQueryTimeout = 60 * time.Second
)
var (
errSQLFoldersAssociation = errors.New("unable to associate virtual folders to user")
errSQLGroupsAssociation = errors.New("unable to associate groups to user")
errSQLUsersAssociation = errors.New("unable to associate users to group")
errSchemaVersionEmpty = errors.New("we can't determine schema version because the schema_migration table is empty. The SFTPGo database might be corrupted. Consider using the \"resetprovider\" sub-command")
)
type sqlQuerier interface {
QueryRowContext(ctx context.Context, query string, args ...any) *sql.Row
QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error)
ExecContext(ctx context.Context, query string, args ...any) (sql.Result, error)
PrepareContext(ctx context.Context, query string) (*sql.Stmt, error)
}
type sqlScanner interface {
Scan(dest ...any) error
}
func sqlReplaceAll(sql string) string {
sql = strings.ReplaceAll(sql, "{{schema_version}}", sqlTableSchemaVersion)
sql = strings.ReplaceAll(sql, "{{admins}}", sqlTableAdmins)
sql = strings.ReplaceAll(sql, "{{folders}}", sqlTableFolders)
sql = strings.ReplaceAll(sql, "{{users}}", sqlTableUsers)
sql = strings.ReplaceAll(sql, "{{groups}}", sqlTableGroups)
sql = strings.ReplaceAll(sql, "{{users_folders_mapping}}", sqlTableUsersFoldersMapping)
sql = strings.ReplaceAll(sql, "{{users_groups_mapping}}", sqlTableUsersGroupsMapping)
sql = strings.ReplaceAll(sql, "{{admins_groups_mapping}}", sqlTableAdminsGroupsMapping)
sql = strings.ReplaceAll(sql, "{{groups_folders_mapping}}", sqlTableGroupsFoldersMapping)
sql = strings.ReplaceAll(sql, "{{api_keys}}", sqlTableAPIKeys)
sql = strings.ReplaceAll(sql, "{{shares}}", sqlTableShares)
sql = strings.ReplaceAll(sql, "{{defender_events}}", sqlTableDefenderEvents)
sql = strings.ReplaceAll(sql, "{{defender_hosts}}", sqlTableDefenderHosts)
sql = strings.ReplaceAll(sql, "{{active_transfers}}", sqlTableActiveTransfers)
sql = strings.ReplaceAll(sql, "{{shared_sessions}}", sqlTableSharedSessions)
sql = strings.ReplaceAll(sql, "{{events_actions}}", sqlTableEventsActions)
sql = strings.ReplaceAll(sql, "{{events_rules}}", sqlTableEventsRules)
sql = strings.ReplaceAll(sql, "{{rules_actions_mapping}}", sqlTableRulesActionsMapping)
sql = strings.ReplaceAll(sql, "{{tasks}}", sqlTableTasks)
sql = strings.ReplaceAll(sql, "{{nodes}}", sqlTableNodes)
sql = strings.ReplaceAll(sql, "{{roles}}", sqlTableRoles)
sql = strings.ReplaceAll(sql, "{{ip_lists}}", sqlTableIPLists)
sql = strings.ReplaceAll(sql, "{{configs}}", sqlTableConfigs)
sql = strings.ReplaceAll(sql, "{{prefix}}", config.SQLTablesPrefix)
return sql
}
func sqlCommonGetShareByID(shareID, username string, dbHandle sqlQuerier) (Share, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
filterUser := username != ""
q := getShareByIDQuery(filterUser)
var row *sql.Row
if filterUser {
row = dbHandle.QueryRowContext(ctx, q, shareID, username)
} else {
row = dbHandle.QueryRowContext(ctx, q, shareID)
}
return getShareFromDbRow(row)
}
func sqlCommonAddShare(share *Share, dbHandle *sql.DB) error {
err := share.validate()
if err != nil {
return err
}
user, err := provider.userExists(share.Username, "")
if err != nil {
return util.NewGenericError(fmt.Sprintf("unable to validate user %q", share.Username))
}
paths, err := json.Marshal(share.Paths)
if err != nil {
return err
}
var allowFrom []byte
if len(share.AllowFrom) > 0 {
res, err := json.Marshal(share.AllowFrom)
if err == nil {
allowFrom = res
}
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAddShareQuery()
usedTokens := 0
createdAt := util.GetTimeAsMsSinceEpoch(time.Now())
updatedAt := createdAt
lastUseAt := int64(0)
if share.IsRestore {
usedTokens = share.UsedTokens
if share.CreatedAt > 0 {
createdAt = share.CreatedAt
}
if share.UpdatedAt > 0 {
updatedAt = share.UpdatedAt
}
lastUseAt = share.LastUseAt
}
_, err = dbHandle.ExecContext(ctx, q, share.ShareID, share.Name, share.Description, share.Scope,
paths, createdAt, updatedAt, lastUseAt, share.ExpiresAt, share.Password,
share.MaxTokens, usedTokens, allowFrom, user.ID)
return err
}
func sqlCommonUpdateShare(share *Share, dbHandle *sql.DB) error {
err := share.validate()
if err != nil {
return err
}
paths, err := json.Marshal(share.Paths)
if err != nil {
return err
}
var allowFrom []byte
if len(share.AllowFrom) > 0 {
res, err := json.Marshal(share.AllowFrom)
if err == nil {
allowFrom = res
}
}
user, err := provider.userExists(share.Username, "")
if err != nil {
return util.NewGenericError(fmt.Sprintf("unable to validate user %q", share.Username))
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
var q string
if share.IsRestore {
q = getUpdateShareRestoreQuery()
} else {
q = getUpdateShareQuery()
}
var res sql.Result
if share.IsRestore {
if share.CreatedAt == 0 {
share.CreatedAt = util.GetTimeAsMsSinceEpoch(time.Now())
}
if share.UpdatedAt == 0 {
share.UpdatedAt = share.CreatedAt
}
res, err = dbHandle.ExecContext(ctx, q, share.Name, share.Description, share.Scope, paths,
share.CreatedAt, share.UpdatedAt, share.LastUseAt, share.ExpiresAt, share.Password, share.MaxTokens,
share.UsedTokens, allowFrom, user.ID, share.ShareID)
} else {
res, err = dbHandle.ExecContext(ctx, q, share.Name, share.Description, share.Scope, paths,
util.GetTimeAsMsSinceEpoch(time.Now()), share.ExpiresAt, share.Password, share.MaxTokens,
allowFrom, user.ID, share.ShareID)
}
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonDeleteShare(share Share, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDeleteShareQuery()
res, err := dbHandle.ExecContext(ctx, q, share.ShareID)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonGetShares(limit, offset int, order, username string, dbHandle sqlQuerier) ([]Share, error) {
shares := make([]Share, 0, limit)
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getSharesQuery(order)
rows, err := dbHandle.QueryContext(ctx, q, username, limit, offset)
if err != nil {
return shares, err
}
defer rows.Close()
for rows.Next() {
s, err := getShareFromDbRow(rows)
if err != nil {
return shares, err
}
s.HideConfidentialData()
shares = append(shares, s)
}
return shares, rows.Err()
}
func sqlCommonDumpShares(dbHandle sqlQuerier) ([]Share, error) {
shares := make([]Share, 0, 30)
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDumpSharesQuery()
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return shares, err
}
defer rows.Close()
for rows.Next() {
s, err := getShareFromDbRow(rows)
if err != nil {
return shares, err
}
shares = append(shares, s)
}
return shares, rows.Err()
}
func sqlCommonGetAPIKeyByID(keyID string, dbHandle sqlQuerier) (APIKey, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAPIKeyByIDQuery()
row := dbHandle.QueryRowContext(ctx, q, keyID)
apiKey, err := getAPIKeyFromDbRow(row)
if err != nil {
return apiKey, err
}
return getAPIKeyWithRelatedFields(ctx, apiKey, dbHandle)
}
func sqlCommonAddAPIKey(apiKey *APIKey, dbHandle *sql.DB) error {
err := apiKey.validate()
if err != nil {
return err
}
userID, adminID, err := sqlCommonGetAPIKeyRelatedIDs(apiKey)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAddAPIKeyQuery()
_, err = dbHandle.ExecContext(ctx, q, apiKey.KeyID, apiKey.Name, apiKey.Key, apiKey.Scope,
util.GetTimeAsMsSinceEpoch(time.Now()), util.GetTimeAsMsSinceEpoch(time.Now()), apiKey.LastUseAt,
apiKey.ExpiresAt, apiKey.Description, userID, adminID)
return err
}
func sqlCommonUpdateAPIKey(apiKey *APIKey, dbHandle *sql.DB) error {
err := apiKey.validate()
if err != nil {
return err
}
userID, adminID, err := sqlCommonGetAPIKeyRelatedIDs(apiKey)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateAPIKeyQuery()
res, err := dbHandle.ExecContext(ctx, q, apiKey.Name, apiKey.Scope, apiKey.ExpiresAt, userID, adminID,
apiKey.Description, util.GetTimeAsMsSinceEpoch(time.Now()), apiKey.KeyID)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonDeleteAPIKey(apiKey APIKey, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDeleteAPIKeyQuery()
res, err := dbHandle.ExecContext(ctx, q, apiKey.KeyID)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonGetAPIKeys(limit, offset int, order string, dbHandle sqlQuerier) ([]APIKey, error) {
apiKeys := make([]APIKey, 0, limit)
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAPIKeysQuery(order)
rows, err := dbHandle.QueryContext(ctx, q, limit, offset)
if err != nil {
return apiKeys, err
}
defer rows.Close()
for rows.Next() {
k, err := getAPIKeyFromDbRow(rows)
if err != nil {
return apiKeys, err
}
k.HideConfidentialData()
apiKeys = append(apiKeys, k)
}
err = rows.Err()
if err != nil {
return apiKeys, err
}
apiKeys, err = getRelatedValuesForAPIKeys(ctx, apiKeys, dbHandle, APIKeyScopeAdmin)
if err != nil {
return apiKeys, err
}
return getRelatedValuesForAPIKeys(ctx, apiKeys, dbHandle, APIKeyScopeUser)
}
func sqlCommonDumpAPIKeys(dbHandle sqlQuerier) ([]APIKey, error) {
apiKeys := make([]APIKey, 0, 30)
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDumpAPIKeysQuery()
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return apiKeys, err
}
defer rows.Close()
for rows.Next() {
k, err := getAPIKeyFromDbRow(rows)
if err != nil {
return apiKeys, err
}
apiKeys = append(apiKeys, k)
}
err = rows.Err()
if err != nil {
return apiKeys, err
}
apiKeys, err = getRelatedValuesForAPIKeys(ctx, apiKeys, dbHandle, APIKeyScopeAdmin)
if err != nil {
return apiKeys, err
}
return getRelatedValuesForAPIKeys(ctx, apiKeys, dbHandle, APIKeyScopeUser)
}
func sqlCommonGetAdminByUsername(username string, dbHandle sqlQuerier) (Admin, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAdminByUsernameQuery()
row := dbHandle.QueryRowContext(ctx, q, username)
admin, err := getAdminFromDbRow(row)
if err != nil {
return admin, err
}
return getAdminWithGroups(ctx, admin, dbHandle)
}
func sqlCommonValidateAdminAndPass(username, password, ip string, dbHandle *sql.DB) (Admin, error) {
admin, err := sqlCommonGetAdminByUsername(username, dbHandle)
if err != nil {
providerLog(logger.LevelWarn, "error authenticating admin %q: %v", username, err)
return admin, err
}
err = admin.checkUserAndPass(password, ip)
return admin, err
}
func sqlCommonAddAdmin(admin *Admin, dbHandle *sql.DB) error {
err := admin.validate()
if err != nil {
return err
}
perms, err := json.Marshal(admin.Permissions)
if err != nil {
return err
}
filters, err := json.Marshal(admin.Filters)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
q := getAddAdminQuery(admin.Role)
_, err = tx.ExecContext(ctx, q, admin.Username, admin.Password, admin.Status, admin.Email, perms,
filters, admin.AdditionalInfo, admin.Description, util.GetTimeAsMsSinceEpoch(time.Now()),
util.GetTimeAsMsSinceEpoch(time.Now()), admin.Role)
if err != nil {
return err
}
return generateAdminGroupMapping(ctx, admin, tx)
})
}
func sqlCommonUpdateAdmin(admin *Admin, dbHandle *sql.DB) error {
err := admin.validate()
if err != nil {
return err
}
perms, err := json.Marshal(admin.Permissions)
if err != nil {
return err
}
filters, err := json.Marshal(admin.Filters)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
q := getUpdateAdminQuery(admin.Role)
_, err = tx.ExecContext(ctx, q, admin.Password, admin.Status, admin.Email, perms, filters,
admin.AdditionalInfo, admin.Description, util.GetTimeAsMsSinceEpoch(time.Now()), admin.Role, admin.Username)
if err != nil {
return err
}
return generateAdminGroupMapping(ctx, admin, tx)
})
}
func sqlCommonDeleteAdmin(admin Admin, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDeleteAdminQuery()
res, err := dbHandle.ExecContext(ctx, q, admin.Username)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonGetAdmins(limit, offset int, order string, dbHandle sqlQuerier) ([]Admin, error) {
admins := make([]Admin, 0, limit)
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAdminsQuery(order)
rows, err := dbHandle.QueryContext(ctx, q, limit, offset)
if err != nil {
return admins, err
}
defer rows.Close()
for rows.Next() {
a, err := getAdminFromDbRow(rows)
if err != nil {
return admins, err
}
a.HideConfidentialData()
admins = append(admins, a)
}
err = rows.Err()
if err != nil {
return admins, err
}
return getAdminsWithGroups(ctx, admins, dbHandle)
}
func sqlCommonDumpAdmins(dbHandle sqlQuerier) ([]Admin, error) {
admins := make([]Admin, 0, 30)
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDumpAdminsQuery()
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return admins, err
}
defer rows.Close()
for rows.Next() {
a, err := getAdminFromDbRow(rows)
if err != nil {
return admins, err
}
admins = append(admins, a)
}
err = rows.Err()
if err != nil {
return admins, err
}
return getAdminsWithGroups(ctx, admins, dbHandle)
}
func sqlCommonGetIPListEntry(ipOrNet string, listType IPListType, dbHandle sqlQuerier) (IPListEntry, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getIPListEntryQuery()
row := dbHandle.QueryRowContext(ctx, q, listType, ipOrNet)
return getIPListEntryFromDbRow(row)
}
func sqlCommonDumpIPListEntries(dbHandle *sql.DB) ([]IPListEntry, error) {
count, err := sqlCommonCountIPListEntries(0, dbHandle)
if err != nil {
return nil, err
}
if count > ipListMemoryLimit {
providerLog(logger.LevelInfo, "IP lists excluded from dump, too many entries: %d", count)
return nil, nil
}
entries := make([]IPListEntry, 0, 100)
ctx, cancel := context.WithTimeout(context.Background(), longSQLQueryTimeout)
defer cancel()
q := getDumpListEntriesQuery()
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return entries, err
}
defer rows.Close()
for rows.Next() {
entry, err := getIPListEntryFromDbRow(rows)
if err != nil {
return entries, err
}
entries = append(entries, entry)
}
return entries, rows.Err()
}
func sqlCommonCountIPListEntries(listType IPListType, dbHandle *sql.DB) (int64, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
var q string
var args []any
if listType == 0 {
q = getCountAllIPListEntriesQuery()
} else {
q = getCountIPListEntriesQuery()
args = append(args, listType)
}
var count int64
err := dbHandle.QueryRowContext(ctx, q, args...).Scan(&count)
return count, err
}
func sqlCommonGetIPListEntries(listType IPListType, filter, from, order string, limit int, dbHandle sqlQuerier) ([]IPListEntry, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getIPListEntriesQuery(filter, from, order, limit)
args := []any{listType}
if from != "" {
args = append(args, from)
}
if filter != "" {
args = append(args, filter+"%")
}
if limit > 0 {
args = append(args, limit)
}
entries := make([]IPListEntry, 0, limit)
rows, err := dbHandle.QueryContext(ctx, q, args...)
if err != nil {
return entries, err
}
defer rows.Close()
for rows.Next() {
entry, err := getIPListEntryFromDbRow(rows)
if err != nil {
return entries, err
}
entries = append(entries, entry)
}
return entries, rows.Err()
}
func sqlCommonGetRecentlyUpdatedIPListEntries(after int64, dbHandle sqlQuerier) ([]IPListEntry, error) {
ctx, cancel := context.WithTimeout(context.Background(), longSQLQueryTimeout)
defer cancel()
q := getRecentlyUpdatedIPListQuery()
entries := make([]IPListEntry, 0, 5)
rows, err := dbHandle.QueryContext(ctx, q, after)
if err != nil {
return entries, err
}
defer rows.Close()
for rows.Next() {
entry, err := getIPListEntryFromDbRow(rows)
if err != nil {
return entries, err
}
entries = append(entries, entry)
}
return entries, rows.Err()
}
func sqlCommonGetListEntriesForIP(ip string, listType IPListType, dbHandle sqlQuerier) ([]IPListEntry, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
var rows *sql.Rows
var err error
entries := make([]IPListEntry, 0, 2)
if config.Driver == PGSQLDataProviderName || config.Driver == CockroachDataProviderName {
rows, err = dbHandle.QueryContext(ctx, getIPListEntriesForIPQueryPg(), listType, ip)
if err != nil {
return entries, err
}
} else {
ipAddr, err := netip.ParseAddr(ip)
if err != nil {
return entries, fmt.Errorf("invalid ip address %s", ip)
}
var netType int
var ipBytes []byte
if ipAddr.Is4() || ipAddr.Is4In6() {
netType = ipTypeV4
as4 := ipAddr.As4()
ipBytes = as4[:]
} else {
netType = ipTypeV6
as16 := ipAddr.As16()
ipBytes = as16[:]
}
rows, err = dbHandle.QueryContext(ctx, getIPListEntriesForIPQueryNoPg(), listType, netType, ipBytes)
if err != nil {
return entries, err
}
}
defer rows.Close()
for rows.Next() {
entry, err := getIPListEntryFromDbRow(rows)
if err != nil {
return entries, err
}
entries = append(entries, entry)
}
return entries, rows.Err()
}
func sqlCommonAddIPListEntry(entry *IPListEntry, dbHandle *sql.DB) error {
if err := entry.validate(); err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
var err error
q := getAddIPListEntryQuery()
first := entry.getFirst()
last := entry.getLast()
var netType int
if first.Is4() {
netType = ipTypeV4
} else {
netType = ipTypeV6
}
if config.IsShared == 1 {
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
_, err := tx.ExecContext(ctx, getRemoveSoftDeletedIPListEntryQuery(), entry.Type, entry.IPOrNet)
if err != nil {
return err
}
if config.Driver == PGSQLDataProviderName || config.Driver == CockroachDataProviderName {
_, err = tx.ExecContext(ctx, q, entry.Type, entry.IPOrNet, first.String(), last.String(),
netType, entry.Protocols, entry.Description, entry.Mode, util.GetTimeAsMsSinceEpoch(time.Now()),
util.GetTimeAsMsSinceEpoch(time.Now()))
} else {
_, err = tx.ExecContext(ctx, q, entry.Type, entry.IPOrNet, entry.First, entry.Last,
netType, entry.Protocols, entry.Description, entry.Mode, util.GetTimeAsMsSinceEpoch(time.Now()),
util.GetTimeAsMsSinceEpoch(time.Now()))
}
return err
})
}
if config.Driver == PGSQLDataProviderName || config.Driver == CockroachDataProviderName {
_, err = dbHandle.ExecContext(ctx, q, entry.Type, entry.IPOrNet, first.String(), last.String(),
netType, entry.Protocols, entry.Description, entry.Mode, util.GetTimeAsMsSinceEpoch(time.Now()),
util.GetTimeAsMsSinceEpoch(time.Now()))
} else {
_, err = dbHandle.ExecContext(ctx, q, entry.Type, entry.IPOrNet, entry.First, entry.Last,
netType, entry.Protocols, entry.Description, entry.Mode, util.GetTimeAsMsSinceEpoch(time.Now()),
util.GetTimeAsMsSinceEpoch(time.Now()))
}
return err
}
func sqlCommonUpdateIPListEntry(entry *IPListEntry, dbHandle *sql.DB) error {
if err := entry.validate(); err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateIPListEntryQuery()
res, err := dbHandle.ExecContext(ctx, q, entry.Mode, entry.Protocols, entry.Description,
util.GetTimeAsMsSinceEpoch(time.Now()), entry.Type, entry.IPOrNet)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonDeleteIPListEntry(entry IPListEntry, softDelete bool, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDeleteIPListEntryQuery(softDelete)
var args []any
if softDelete {
ts := util.GetTimeAsMsSinceEpoch(time.Now())
args = append(args, ts, ts)
}
args = append(args, entry.Type, entry.IPOrNet)
res, err := dbHandle.ExecContext(ctx, q, args...)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonGetRoleByName(name string, dbHandle sqlQuerier) (Role, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getRoleByNameQuery()
row := dbHandle.QueryRowContext(ctx, q, name)
role, err := getRoleFromDbRow(row)
if err != nil {
return role, err
}
role, err = getRoleWithUsers(ctx, role, dbHandle)
if err != nil {
return role, err
}
return getRoleWithAdmins(ctx, role, dbHandle)
}
func sqlCommonDumpRoles(dbHandle sqlQuerier) ([]Role, error) {
roles := make([]Role, 0, 10)
ctx, cancel := context.WithTimeout(context.Background(), longSQLQueryTimeout)
defer cancel()
q := getDumpRolesQuery()
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return roles, err
}
defer rows.Close()
for rows.Next() {
role, err := getRoleFromDbRow(rows)
if err != nil {
return roles, err
}
roles = append(roles, role)
}
return roles, rows.Err()
}
func sqlCommonGetRoles(limit int, offset int, order string, minimal bool, dbHandle sqlQuerier) ([]Role, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getRolesQuery(order, minimal)
roles := make([]Role, 0, limit)
rows, err := dbHandle.QueryContext(ctx, q, limit, offset)
if err != nil {
return roles, err
}
defer rows.Close()
for rows.Next() {
var role Role
if minimal {
err = rows.Scan(&role.ID, &role.Name)
} else {
role, err = getRoleFromDbRow(rows)
}
if err != nil {
return roles, err
}
roles = append(roles, role)
}
err = rows.Err()
if err != nil {
return roles, err
}
if minimal {
return roles, nil
}
roles, err = getRolesWithUsers(ctx, roles, dbHandle)
if err != nil {
return roles, err
}
return getRolesWithAdmins(ctx, roles, dbHandle)
}
func sqlCommonAddRole(role *Role, dbHandle *sql.DB) error {
if err := role.validate(); err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAddRoleQuery()
_, err := dbHandle.ExecContext(ctx, q, role.Name, role.Description, util.GetTimeAsMsSinceEpoch(time.Now()),
util.GetTimeAsMsSinceEpoch(time.Now()))
return err
}
func sqlCommonUpdateRole(role *Role, dbHandle *sql.DB) error {
if err := role.validate(); err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateRoleQuery()
res, err := dbHandle.ExecContext(ctx, q, role.Description, util.GetTimeAsMsSinceEpoch(time.Now()), role.Name)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonDeleteRole(role Role, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDeleteRoleQuery()
res, err := dbHandle.ExecContext(ctx, q, role.Name)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonGetGroupByName(name string, dbHandle sqlQuerier) (Group, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getGroupByNameQuery()
row := dbHandle.QueryRowContext(ctx, q, name)
group, err := getGroupFromDbRow(row)
if err != nil {
return group, err
}
group, err = getGroupWithVirtualFolders(ctx, group, dbHandle)
if err != nil {
return group, err
}
group, err = getGroupWithUsers(ctx, group, dbHandle)
if err != nil {
return group, err
}
return getGroupWithAdmins(ctx, group, dbHandle)
}
func sqlCommonDumpGroups(dbHandle sqlQuerier) ([]Group, error) {
groups := make([]Group, 0, 50)
ctx, cancel := context.WithTimeout(context.Background(), longSQLQueryTimeout)
defer cancel()
q := getDumpGroupsQuery()
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return groups, err
}
defer rows.Close()
for rows.Next() {
group, err := getGroupFromDbRow(rows)
if err != nil {
return groups, err
}
groups = append(groups, group)
}
err = rows.Err()
if err != nil {
return groups, err
}
return getGroupsWithVirtualFolders(ctx, groups, dbHandle)
}
func sqlCommonGetUsersInGroups(names []string, dbHandle sqlQuerier) ([]string, error) {
if len(names) == 0 {
return nil, nil
}
maxNames := len(sqlPlaceholders)
usernames := make([]string, 0, len(names))
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
for len(names) > 0 {
if maxNames > len(names) {
maxNames = len(names)
}
q := getUsersInGroupsQuery(maxNames)
args := make([]any, 0, maxNames)
for _, name := range names[:maxNames] {
args = append(args, name)
}
rows, err := dbHandle.QueryContext(ctx, q, args...)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var username string
err = rows.Scan(&username)
if err != nil {
return usernames, err
}
usernames = append(usernames, username)
}
err = rows.Err()
if err != nil {
return usernames, err
}
names = names[maxNames:]
}
return usernames, nil
}
func sqlCommonGetGroupsWithNames(names []string, dbHandle sqlQuerier) ([]Group, error) {
if len(names) == 0 {
return nil, nil
}
maxNames := len(sqlPlaceholders)
groups := make([]Group, 0, len(names))
for len(names) > 0 {
if maxNames > len(names) {
maxNames = len(names)
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getGroupsWithNamesQuery(maxNames)
args := make([]any, 0, maxNames)
for _, name := range names[:maxNames] {
args = append(args, name)
}
rows, err := dbHandle.QueryContext(ctx, q, args...)
if err != nil {
return groups, err
}
defer rows.Close()
for rows.Next() {
group, err := getGroupFromDbRow(rows)
if err != nil {
return groups, err
}
groups = append(groups, group)
}
err = rows.Err()
if err != nil {
return groups, err
}
names = names[maxNames:]
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
return getGroupsWithVirtualFolders(ctx, groups, dbHandle)
}
func sqlCommonGetGroups(limit int, offset int, order string, minimal bool, dbHandle sqlQuerier) ([]Group, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getGroupsQuery(order, minimal)
groups := make([]Group, 0, limit)
rows, err := dbHandle.QueryContext(ctx, q, limit, offset)
if err != nil {
return groups, err
}
defer rows.Close()
for rows.Next() {
var group Group
if minimal {
err = rows.Scan(&group.ID, &group.Name)
} else {
group, err = getGroupFromDbRow(rows)
}
if err != nil {
return groups, err
}
groups = append(groups, group)
}
err = rows.Err()
if err != nil {
return groups, err
}
if minimal {
return groups, nil
}
groups, err = getGroupsWithVirtualFolders(ctx, groups, dbHandle)
if err != nil {
return groups, err
}
groups, err = getGroupsWithUsers(ctx, groups, dbHandle)
if err != nil {
return groups, err
}
groups, err = getGroupsWithAdmins(ctx, groups, dbHandle)
if err != nil {
return groups, err
}
for idx := range groups {
groups[idx].PrepareForRendering()
}
return groups, nil
}
func sqlCommonAddGroup(group *Group, dbHandle *sql.DB) error {
if err := group.validate(); err != nil {
return err
}
settings, err := json.Marshal(group.UserSettings)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
q := getAddGroupQuery()
_, err := tx.ExecContext(ctx, q, group.Name, group.Description, util.GetTimeAsMsSinceEpoch(time.Now()),
util.GetTimeAsMsSinceEpoch(time.Now()), settings)
if err != nil {
return err
}
return generateGroupVirtualFoldersMapping(ctx, group, tx)
})
}
func sqlCommonUpdateGroup(group *Group, dbHandle *sql.DB) error {
if err := group.validate(); err != nil {
return err
}
settings, err := json.Marshal(group.UserSettings)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
q := getUpdateGroupQuery()
_, err := tx.ExecContext(ctx, q, group.Description, settings, util.GetTimeAsMsSinceEpoch(time.Now()), group.Name)
if err != nil {
return err
}
return generateGroupVirtualFoldersMapping(ctx, group, tx)
})
}
func sqlCommonDeleteGroup(group Group, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDeleteGroupQuery()
res, err := dbHandle.ExecContext(ctx, q, group.Name)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonGetUserByUsername(username, role string, dbHandle sqlQuerier) (User, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUserByUsernameQuery(role)
args := []any{username}
if role != "" {
args = append(args, role)
}
row := dbHandle.QueryRowContext(ctx, q, args...)
user, err := getUserFromDbRow(row)
if err != nil {
return user, err
}
user, err = getUserWithVirtualFolders(ctx, user, dbHandle)
if err != nil {
return user, err
}
return getUserWithGroups(ctx, user, dbHandle)
}
func sqlCommonValidateUserAndPass(username, password, ip, protocol string, dbHandle *sql.DB) (User, error) {
user, err := sqlCommonGetUserByUsername(username, "", dbHandle)
if err != nil {
providerLog(logger.LevelWarn, "error authenticating user %q: %v", username, err)
return user, err
}
return checkUserAndPass(&user, password, ip, protocol)
}
func sqlCommonValidateUserAndTLSCertificate(username, protocol string, tlsCert *x509.Certificate, dbHandle *sql.DB) (User, error) {
var user User
if tlsCert == nil {
return user, errors.New("TLS certificate cannot be null or empty")
}
user, err := sqlCommonGetUserByUsername(username, "", dbHandle)
if err != nil {
providerLog(logger.LevelWarn, "error authenticating user %q: %v", username, err)
return user, err
}
return checkUserAndTLSCertificate(&user, protocol, tlsCert)
}
func sqlCommonValidateUserAndPubKey(username string, pubKey []byte, isSSHCert bool, dbHandle *sql.DB) (User, string, error) {
var user User
if len(pubKey) == 0 {
return user, "", errors.New("credentials cannot be null or empty")
}
user, err := sqlCommonGetUserByUsername(username, "", dbHandle)
if err != nil {
providerLog(logger.LevelWarn, "error authenticating user %q: %v", username, err)
return user, "", err
}
return checkUserAndPubKey(&user, pubKey, isSSHCert)
}
func sqlCommonCheckAvailability(dbHandle *sql.DB) (err error) {
defer func() {
if r := recover(); r != nil {
providerLog(logger.LevelError, "panic in check provider availability, stack trace: %s", string(debug.Stack()))
err = errors.New("unable to check provider status")
}
}()
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
err = dbHandle.PingContext(ctx)
return
}
func sqlCommonUpdateTransferQuota(username string, uploadSize, downloadSize int64, reset bool, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateTransferQuotaQuery(reset)
_, err := dbHandle.ExecContext(ctx, q, uploadSize, downloadSize, util.GetTimeAsMsSinceEpoch(time.Now()), username)
if err == nil {
providerLog(logger.LevelDebug, "transfer quota updated for user %q, ul increment: %d dl increment: %d is reset? %t",
username, uploadSize, downloadSize, reset)
} else {
providerLog(logger.LevelError, "error updating quota for user %q: %v", username, err)
}
return err
}
func sqlCommonUpdateQuota(username string, filesAdd int, sizeAdd int64, reset bool, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateQuotaQuery(reset)
_, err := dbHandle.ExecContext(ctx, q, sizeAdd, filesAdd, util.GetTimeAsMsSinceEpoch(time.Now()), username)
if err == nil {
providerLog(logger.LevelDebug, "quota updated for user %q, files increment: %d size increment: %d is reset? %t",
username, filesAdd, sizeAdd, reset)
} else {
providerLog(logger.LevelError, "error updating quota for user %q: %v", username, err)
}
return err
}
func sqlCommonGetUsedQuota(username string, dbHandle *sql.DB) (int, int64, int64, int64, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getQuotaQuery()
var usedFiles int
var usedSize, usedUploadSize, usedDownloadSize int64
err := dbHandle.QueryRowContext(ctx, q, username).Scan(&usedSize, &usedFiles, &usedUploadSize, &usedDownloadSize)
if err != nil {
providerLog(logger.LevelError, "error getting quota for user: %v, error: %v", username, err)
return 0, 0, 0, 0, err
}
return usedFiles, usedSize, usedUploadSize, usedDownloadSize, err
}
func sqlCommonUpdateShareLastUse(shareID string, numTokens int, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateShareLastUseQuery()
_, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), numTokens, shareID)
if err == nil {
providerLog(logger.LevelDebug, "last use updated for shared object %q", shareID)
} else {
providerLog(logger.LevelWarn, "error updating last use for shared object %q: %v", shareID, err)
}
return err
}
func sqlCommonUpdateAPIKeyLastUse(keyID string, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateAPIKeyLastUseQuery()
_, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), keyID)
if err == nil {
providerLog(logger.LevelDebug, "last use updated for key %q", keyID)
} else {
providerLog(logger.LevelWarn, "error updating last use for key %q: %v", keyID, err)
}
return err
}
func sqlCommonUpdateAdminLastLogin(username string, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateAdminLastLoginQuery()
_, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), username)
if err == nil {
providerLog(logger.LevelDebug, "last login updated for admin %q", username)
} else {
providerLog(logger.LevelWarn, "error updating last login for admin %q: %v", username, err)
}
return err
}
func sqlCommonSetUpdatedAt(username string, dbHandle *sql.DB) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getSetUpdateAtQuery()
_, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), username)
if err == nil {
providerLog(logger.LevelDebug, "updated_at set for user %q", username)
} else {
providerLog(logger.LevelWarn, "error setting updated_at for user %q: %v", username, err)
}
}
func sqlCommonSetFirstDownloadTimestamp(username string, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getSetFirstDownloadQuery()
res, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), username)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonSetFirstUploadTimestamp(username string, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getSetFirstUploadQuery()
res, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), username)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonUpdateLastLogin(username string, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateLastLoginQuery()
_, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), username)
if err == nil {
providerLog(logger.LevelDebug, "last login updated for user %q", username)
} else {
providerLog(logger.LevelWarn, "error updating last login for user %q: %v", username, err)
}
return err
}
func sqlCommonAddUser(user *User, dbHandle *sql.DB) error {
err := ValidateUser(user)
if err != nil {
return err
}
permissions, err := user.GetPermissionsAsJSON()
if err != nil {
return err
}
publicKeys, err := user.GetPublicKeysAsJSON()
if err != nil {
return err
}
filters, err := user.GetFiltersAsJSON()
if err != nil {
return err
}
fsConfig, err := user.GetFsConfigAsJSON()
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
if config.IsShared == 1 {
_, err := tx.ExecContext(ctx, getRemoveSoftDeletedUserQuery(), user.Username)
if err != nil {
return err
}
}
q := getAddUserQuery(user.Role)
_, err := tx.ExecContext(ctx, q, user.Username, user.Password, publicKeys, user.HomeDir, user.UID, user.GID,
user.MaxSessions, user.QuotaSize, user.QuotaFiles, permissions, user.UploadBandwidth,
user.DownloadBandwidth, user.Status, user.ExpirationDate, filters, fsConfig, user.AdditionalInfo,
user.Description, user.Email, util.GetTimeAsMsSinceEpoch(time.Now()), util.GetTimeAsMsSinceEpoch(time.Now()),
user.UploadDataTransfer, user.DownloadDataTransfer, user.TotalDataTransfer, user.Role, user.LastPasswordChange)
if err != nil {
return err
}
if err := generateUserVirtualFoldersMapping(ctx, user, tx); err != nil {
return err
}
return generateUserGroupMapping(ctx, user, tx)
})
}
func sqlCommonUpdateUserPassword(username, password string, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateUserPasswordQuery()
res, err := dbHandle.ExecContext(ctx, q, password, util.GetTimeAsMsSinceEpoch(time.Now()), username)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonUpdateUser(user *User, dbHandle *sql.DB) error {
err := ValidateUser(user)
if err != nil {
return err
}
permissions, err := user.GetPermissionsAsJSON()
if err != nil {
return err
}
publicKeys, err := user.GetPublicKeysAsJSON()
if err != nil {
return err
}
filters, err := user.GetFiltersAsJSON()
if err != nil {
return err
}
fsConfig, err := user.GetFsConfigAsJSON()
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
q := getUpdateUserQuery(user.Role)
res, err := tx.ExecContext(ctx, q, user.Password, publicKeys, user.HomeDir, user.UID, user.GID, user.MaxSessions,
user.QuotaSize, user.QuotaFiles, permissions, user.UploadBandwidth, user.DownloadBandwidth, user.Status,
user.ExpirationDate, filters, fsConfig, user.AdditionalInfo, user.Description, user.Email,
util.GetTimeAsMsSinceEpoch(time.Now()), user.UploadDataTransfer, user.DownloadDataTransfer, user.TotalDataTransfer,
user.Role, user.LastPasswordChange, user.Username)
if err != nil {
return err
}
if err := sqlCommonRequireRowAffected(res); err != nil {
return err
}
if err := generateUserVirtualFoldersMapping(ctx, user, tx); err != nil {
return err
}
return generateUserGroupMapping(ctx, user, tx)
})
}
func sqlCommonDeleteUser(user User, softDelete bool, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDeleteUserQuery(softDelete)
if softDelete {
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
if err := sqlCommonClearUserFolderMapping(ctx, &user, tx); err != nil {
return err
}
if err := sqlCommonClearUserGroupMapping(ctx, &user, tx); err != nil {
return err
}
ts := util.GetTimeAsMsSinceEpoch(time.Now())
res, err := tx.ExecContext(ctx, q, ts, ts, user.Username)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
})
}
res, err := dbHandle.ExecContext(ctx, q, user.Username)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonDumpUsers(dbHandle sqlQuerier) ([]User, error) {
users := make([]User, 0, 100)
ctx, cancel := context.WithTimeout(context.Background(), longSQLQueryTimeout)
defer cancel()
q := getDumpUsersQuery()
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return users, err
}
defer rows.Close()
for rows.Next() {
u, err := getUserFromDbRow(rows)
if err != nil {
return users, err
}
users = append(users, u)
}
err = rows.Err()
if err != nil {
return users, err
}
users, err = getUsersWithVirtualFolders(ctx, users, dbHandle)
if err != nil {
return users, err
}
return getUsersWithGroups(ctx, users, dbHandle)
}
func sqlCommonGetRecentlyUpdatedUsers(after int64, dbHandle sqlQuerier) ([]User, error) {
users := make([]User, 0, 10)
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getRecentlyUpdatedUsersQuery()
rows, err := dbHandle.QueryContext(ctx, q, after)
if err != nil {
return users, err
}
defer rows.Close()
for rows.Next() {
u, err := getUserFromDbRow(rows)
if err != nil {
return users, err
}
users = append(users, u)
}
err = rows.Err()
if err != nil {
return users, err
}
users, err = getUsersWithVirtualFolders(ctx, users, dbHandle)
if err != nil {
return users, err
}
users, err = getUsersWithGroups(ctx, users, dbHandle)
if err != nil {
return users, err
}
var groupNames []string
for _, u := range users {
for _, g := range u.Groups {
groupNames = append(groupNames, g.Name)
}
}
groupNames = util.RemoveDuplicates(groupNames, false)
if len(groupNames) == 0 {
return users, nil
}
groups, err := sqlCommonGetGroupsWithNames(groupNames, dbHandle)
if err != nil {
return users, err
}
if len(groups) == 0 {
return users, nil
}
groupsMapping := make(map[string]Group)
for idx := range groups {
groupsMapping[groups[idx].Name] = groups[idx]
}
for idx := range users {
ref := &users[idx]
ref.applyGroupSettings(groupsMapping)
}
return users, nil
}
func sqlGetMaxUsersForQuotaCheckRange() int {
maxUsers := 50
if maxUsers > len(sqlPlaceholders) {
maxUsers = len(sqlPlaceholders)
}
return maxUsers
}
func sqlCommonGetUsersForQuotaCheck(toFetch map[string]bool, dbHandle sqlQuerier) ([]User, error) {
maxUsers := sqlGetMaxUsersForQuotaCheckRange()
users := make([]User, 0, maxUsers)
usernames := make([]string, 0, len(toFetch))
for k := range toFetch {
usernames = append(usernames, k)
}
for len(usernames) > 0 {
if maxUsers > len(usernames) {
maxUsers = len(usernames)
}
usersRange, err := sqlCommonGetUsersRangeForQuotaCheck(usernames[:maxUsers], dbHandle)
if err != nil {
return users, err
}
users = append(users, usersRange...)
usernames = usernames[maxUsers:]
}
var usersWithFolders []User
validIdx := 0
for _, user := range users {
if toFetch[user.Username] {
usersWithFolders = append(usersWithFolders, user)
} else {
users[validIdx] = user
validIdx++
}
}
users = users[:validIdx]
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
usersWithFolders, err := getUsersWithVirtualFolders(ctx, usersWithFolders, dbHandle)
if err != nil {
return users, err
}
users = append(users, usersWithFolders...)
users, err = getUsersWithGroups(ctx, users, dbHandle)
if err != nil {
return users, err
}
var groupNames []string
for _, u := range users {
for _, g := range u.Groups {
groupNames = append(groupNames, g.Name)
}
}
groupNames = util.RemoveDuplicates(groupNames, false)
if len(groupNames) == 0 {
return users, nil
}
groups, err := sqlCommonGetGroupsWithNames(groupNames, dbHandle)
if err != nil {
return users, err
}
groupsMapping := make(map[string]Group)
for idx := range groups {
groupsMapping[groups[idx].Name] = groups[idx]
}
for idx := range users {
ref := &users[idx]
ref.applyGroupSettings(groupsMapping)
}
return users, nil
}
func sqlCommonGetUsersRangeForQuotaCheck(usernames []string, dbHandle sqlQuerier) ([]User, error) {
users := make([]User, 0, len(usernames))
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUsersForQuotaCheckQuery(len(usernames))
queryArgs := make([]any, 0, len(usernames))
for idx := range usernames {
queryArgs = append(queryArgs, usernames[idx])
}
rows, err := dbHandle.QueryContext(ctx, q, queryArgs...)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var user User
var filters []byte
err = rows.Scan(&user.ID, &user.Username, &user.QuotaSize, &user.UsedQuotaSize, &user.TotalDataTransfer,
&user.UploadDataTransfer, &user.DownloadDataTransfer, &user.UsedUploadDataTransfer,
&user.UsedDownloadDataTransfer, &filters)
if err != nil {
return users, err
}
var userFilters UserFilters
err = json.Unmarshal(filters, &userFilters)
if err == nil {
user.Filters = userFilters
}
users = append(users, user)
}
return users, rows.Err()
}
func sqlCommonAddActiveTransfer(transfer ActiveTransfer, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAddActiveTransferQuery()
now := util.GetTimeAsMsSinceEpoch(time.Now())
_, err := dbHandle.ExecContext(ctx, q, transfer.ID, transfer.ConnID, transfer.Type, transfer.Username,
transfer.FolderName, transfer.IP, transfer.TruncatedSize, transfer.CurrentULSize, transfer.CurrentDLSize,
now, now)
return err
}
func sqlCommonUpdateActiveTransferSizes(ulSize, dlSize, transferID int64, connectionID string, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateActiveTransferSizesQuery()
_, err := dbHandle.ExecContext(ctx, q, ulSize, dlSize, util.GetTimeAsMsSinceEpoch(time.Now()), connectionID, transferID)
return err
}
func sqlCommonRemoveActiveTransfer(transferID int64, connectionID string, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getRemoveActiveTransferQuery()
_, err := dbHandle.ExecContext(ctx, q, connectionID, transferID)
return err
}
func sqlCommonCleanupActiveTransfers(before time.Time, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getCleanupActiveTransfersQuery()
_, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(before))
return err
}
func sqlCommonGetActiveTransfers(from time.Time, dbHandle sqlQuerier) ([]ActiveTransfer, error) {
transfers := make([]ActiveTransfer, 0, 30)
ctx, cancel := context.WithTimeout(context.Background(), longSQLQueryTimeout)
defer cancel()
q := getActiveTransfersQuery()
rows, err := dbHandle.QueryContext(ctx, q, util.GetTimeAsMsSinceEpoch(from))
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var transfer ActiveTransfer
var folderName sql.NullString
err = rows.Scan(&transfer.ID, &transfer.ConnID, &transfer.Type, &transfer.Username, &folderName, &transfer.IP,
&transfer.TruncatedSize, &transfer.CurrentULSize, &transfer.CurrentDLSize, &transfer.CreatedAt,
&transfer.UpdatedAt)
if err != nil {
return transfers, err
}
if folderName.Valid {
transfer.FolderName = folderName.String
}
transfers = append(transfers, transfer)
}
return transfers, rows.Err()
}
func sqlCommonGetUsers(limit int, offset int, order, role string, dbHandle sqlQuerier) ([]User, error) {
users := make([]User, 0, limit)
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUsersQuery(order, role)
var args []any
if role == "" {
args = append(args, limit, offset)
} else {
args = append(args, role, limit, offset)
}
rows, err := dbHandle.QueryContext(ctx, q, args...)
if err != nil {
return users, err
}
defer rows.Close()
for rows.Next() {
u, err := getUserFromDbRow(rows)
if err != nil {
return users, err
}
users = append(users, u)
}
err = rows.Err()
if err != nil {
return users, err
}
users, err = getUsersWithVirtualFolders(ctx, users, dbHandle)
if err != nil {
return users, err
}
users, err = getUsersWithGroups(ctx, users, dbHandle)
if err != nil {
return users, err
}
for idx := range users {
users[idx].PrepareForRendering()
}
return users, nil
}
func sqlCommonGetDefenderHosts(from int64, limit int, dbHandle sqlQuerier) ([]DefenderEntry, error) {
hosts := make([]DefenderEntry, 0, 100)
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDefenderHostsQuery()
rows, err := dbHandle.QueryContext(ctx, q, from, limit)
if err != nil {
providerLog(logger.LevelError, "unable to get defender hosts: %v", err)
return hosts, err
}
defer rows.Close()
var idForScores []int64
for rows.Next() {
var banTime sql.NullInt64
host := DefenderEntry{}
err = rows.Scan(&host.ID, &host.IP, &banTime)
if err != nil {
providerLog(logger.LevelError, "unable to scan defender host row: %v", err)
return hosts, err
}
var hostBanTime time.Time
if banTime.Valid && banTime.Int64 > 0 {
hostBanTime = util.GetTimeFromMsecSinceEpoch(banTime.Int64)
}
if hostBanTime.IsZero() || hostBanTime.Before(time.Now()) {
idForScores = append(idForScores, host.ID)
} else {
host.BanTime = hostBanTime
}
hosts = append(hosts, host)
}
err = rows.Err()
if err != nil {
providerLog(logger.LevelError, "unable to iterate over defender host rows: %v", err)
return hosts, err
}
return getDefenderHostsWithScores(ctx, hosts, from, idForScores, dbHandle)
}
func sqlCommonIsDefenderHostBanned(ip string, dbHandle sqlQuerier) (DefenderEntry, error) {
var host DefenderEntry
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDefenderIsHostBannedQuery()
row := dbHandle.QueryRowContext(ctx, q, ip, util.GetTimeAsMsSinceEpoch(time.Now()))
err := row.Scan(&host.ID)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return host, util.NewRecordNotFoundError("host not found")
}
providerLog(logger.LevelError, "unable to check ban status for host %q: %v", ip, err)
return host, err
}
return host, nil
}
func sqlCommonGetDefenderHostByIP(ip string, from int64, dbHandle sqlQuerier) (DefenderEntry, error) {
var host DefenderEntry
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDefenderHostQuery()
row := dbHandle.QueryRowContext(ctx, q, ip, from)
var banTime sql.NullInt64
err := row.Scan(&host.ID, &host.IP, &banTime)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return host, util.NewRecordNotFoundError("host not found")
}
providerLog(logger.LevelError, "unable to get host for ip %q: %v", ip, err)
return host, err
}
if banTime.Valid && banTime.Int64 > 0 {
hostBanTime := util.GetTimeFromMsecSinceEpoch(banTime.Int64)
if !hostBanTime.IsZero() && hostBanTime.After(time.Now()) {
host.BanTime = hostBanTime
return host, nil
}
}
hosts, err := getDefenderHostsWithScores(ctx, []DefenderEntry{host}, from, []int64{host.ID}, dbHandle)
if err != nil {
return host, err
}
if len(hosts) == 0 {
return host, util.NewRecordNotFoundError("host not found")
}
return hosts[0], nil
}
func sqlCommonDefenderIncrementBanTime(ip string, minutesToAdd int, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDefenderIncrementBanTimeQuery()
_, err := dbHandle.ExecContext(ctx, q, minutesToAdd*60000, ip)
if err == nil {
providerLog(logger.LevelDebug, "ban time updated for ip %q, increment (minutes): %v",
ip, minutesToAdd)
} else {
providerLog(logger.LevelError, "error updating ban time for ip %q: %v", ip, err)
}
return err
}
func sqlCommonSetDefenderBanTime(ip string, banTime int64, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDefenderSetBanTimeQuery()
_, err := dbHandle.ExecContext(ctx, q, banTime, ip)
if err == nil {
providerLog(logger.LevelDebug, "ip %q banned until %v", ip, util.GetTimeFromMsecSinceEpoch(banTime))
} else {
providerLog(logger.LevelError, "error setting ban time for ip %q: %v", ip, err)
}
return err
}
func sqlCommonDeleteDefenderHost(ip string, dbHandle sqlQuerier) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDeleteDefenderHostQuery()
res, err := dbHandle.ExecContext(ctx, q, ip)
if err != nil {
providerLog(logger.LevelError, "unable to delete defender host %q: %v", ip, err)
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonAddDefenderHostAndEvent(ip string, score int, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
if err := sqlCommonAddDefenderHost(ctx, ip, tx); err != nil {
return err
}
return sqlCommonAddDefenderEvent(ctx, ip, score, tx)
})
}
func sqlCommonDefenderCleanup(from int64, dbHandler *sql.DB) error {
if err := sqlCommonCleanupDefenderEvents(from, dbHandler); err != nil {
return err
}
return sqlCommonCleanupDefenderHosts(from, dbHandler)
}
func sqlCommonAddDefenderHost(ctx context.Context, ip string, tx *sql.Tx) error {
q := getAddDefenderHostQuery()
_, err := tx.ExecContext(ctx, q, ip, util.GetTimeAsMsSinceEpoch(time.Now()))
if err != nil {
providerLog(logger.LevelError, "unable to add defender host %q: %v", ip, err)
}
return err
}
func sqlCommonAddDefenderEvent(ctx context.Context, ip string, score int, tx *sql.Tx) error {
q := getAddDefenderEventQuery()
_, err := tx.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), score, ip)
if err != nil {
providerLog(logger.LevelError, "unable to add defender event for %q: %v", ip, err)
}
return err
}
func sqlCommonCleanupDefenderHosts(from int64, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDefenderHostsCleanupQuery()
_, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), from)
if err != nil {
providerLog(logger.LevelError, "unable to cleanup defender hosts: %v", err)
}
return err
}
func sqlCommonCleanupDefenderEvents(from int64, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDefenderEventsCleanupQuery()
_, err := dbHandle.ExecContext(ctx, q, from)
if err != nil {
providerLog(logger.LevelError, "unable to cleanup defender events: %v", err)
}
return err
}
func getShareFromDbRow(row sqlScanner) (Share, error) {
var share Share
var description, password sql.NullString
var allowFrom, paths []byte
err := row.Scan(&share.ShareID, &share.Name, &description, &share.Scope,
&paths, &share.Username, &share.CreatedAt, &share.UpdatedAt,
&share.LastUseAt, &share.ExpiresAt, &password, &share.MaxTokens,
&share.UsedTokens, &allowFrom)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return share, util.NewRecordNotFoundError(err.Error())
}
return share, err
}
var list []string
err = json.Unmarshal(paths, &list)
if err != nil {
return share, err
}
share.Paths = list
if description.Valid {
share.Description = description.String
}
if password.Valid {
share.Password = password.String
}
list = nil
err = json.Unmarshal(allowFrom, &list)
if err == nil {
share.AllowFrom = list
}
return share, nil
}
func getAPIKeyFromDbRow(row sqlScanner) (APIKey, error) {
var apiKey APIKey
var userID, adminID sql.NullInt64
var description sql.NullString
err := row.Scan(&apiKey.KeyID, &apiKey.Name, &apiKey.Key, &apiKey.Scope, &apiKey.CreatedAt, &apiKey.UpdatedAt,
&apiKey.LastUseAt, &apiKey.ExpiresAt, &description, &userID, &adminID)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return apiKey, util.NewRecordNotFoundError(err.Error())
}
return apiKey, err
}
if userID.Valid {
apiKey.userID = userID.Int64
}
if adminID.Valid {
apiKey.adminID = adminID.Int64
}
if description.Valid {
apiKey.Description = description.String
}
return apiKey, nil
}
func getAdminFromDbRow(row sqlScanner) (Admin, error) {
var admin Admin
var email, additionalInfo, description, role sql.NullString
var permissions, filters []byte
err := row.Scan(&admin.ID, &admin.Username, &admin.Password, &admin.Status, &email, &permissions,
&filters, &additionalInfo, &description, &admin.CreatedAt, &admin.UpdatedAt, &admin.LastLogin, &role)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return admin, util.NewRecordNotFoundError(err.Error())
}
return admin, err
}
var perms []string
err = json.Unmarshal(permissions, &perms)
if err != nil {
return admin, err
}
admin.Permissions = perms
if email.Valid {
admin.Email = email.String
}
var adminFilters AdminFilters
err = json.Unmarshal(filters, &adminFilters)
if err == nil {
admin.Filters = adminFilters
}
if additionalInfo.Valid {
admin.AdditionalInfo = additionalInfo.String
}
if description.Valid {
admin.Description = description.String
}
if role.Valid {
admin.Role = role.String
}
admin.SetEmptySecretsIfNil()
return admin, nil
}
func getEventActionFromDbRow(row sqlScanner) (BaseEventAction, error) {
var action BaseEventAction
var description sql.NullString
var options []byte
err := row.Scan(&action.ID, &action.Name, &description, &action.Type, &options)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return action, util.NewRecordNotFoundError(err.Error())
}
return action, err
}
if description.Valid {
action.Description = description.String
}
var actionOptions BaseEventActionOptions
err = json.Unmarshal(options, &actionOptions)
if err == nil {
action.Options = actionOptions
}
return action, nil
}
func getEventRuleFromDbRow(row sqlScanner) (EventRule, error) {
var rule EventRule
var description sql.NullString
var conditions []byte
err := row.Scan(&rule.ID, &rule.Name, &description, &rule.CreatedAt, &rule.UpdatedAt, &rule.Trigger,
&conditions, &rule.DeletedAt, &rule.Status)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return rule, util.NewRecordNotFoundError(err.Error())
}
return rule, err
}
var ruleConditions EventConditions
err = json.Unmarshal(conditions, &ruleConditions)
if err == nil {
rule.Conditions = ruleConditions
}
if description.Valid {
rule.Description = description.String
}
return rule, nil
}
func getIPListEntryFromDbRow(row sqlScanner) (IPListEntry, error) {
var entry IPListEntry
var description sql.NullString
err := row.Scan(&entry.Type, &entry.IPOrNet, &entry.Mode, &entry.Protocols, &description,
&entry.CreatedAt, &entry.UpdatedAt, &entry.DeletedAt)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return entry, util.NewRecordNotFoundError(err.Error())
}
return entry, err
}
if description.Valid {
entry.Description = description.String
}
return entry, err
}
func getRoleFromDbRow(row sqlScanner) (Role, error) {
var role Role
var description sql.NullString
err := row.Scan(&role.ID, &role.Name, &description, &role.CreatedAt, &role.UpdatedAt)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return role, util.NewRecordNotFoundError(err.Error())
}
return role, err
}
if description.Valid {
role.Description = description.String
}
return role, nil
}
func getGroupFromDbRow(row sqlScanner) (Group, error) {
var group Group
var description sql.NullString
var userSettings []byte
err := row.Scan(&group.ID, &group.Name, &description, &group.CreatedAt, &group.UpdatedAt, &userSettings)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return group, util.NewRecordNotFoundError(err.Error())
}
return group, err
}
if description.Valid {
group.Description = description.String
}
var settings GroupUserSettings
err = json.Unmarshal(userSettings, &settings)
if err == nil {
group.UserSettings = settings
}
return group, nil
}
func getUserFromDbRow(row sqlScanner) (User, error) {
var user User
var password sql.NullString
var permissions, publicKey, filters, fsConfig []byte
var additionalInfo, description, email, role sql.NullString
err := row.Scan(&user.ID, &user.Username, &password, &publicKey, &user.HomeDir, &user.UID, &user.GID, &user.MaxSessions,
&user.QuotaSize, &user.QuotaFiles, &permissions, &user.UsedQuotaSize, &user.UsedQuotaFiles, &user.LastQuotaUpdate,
&user.UploadBandwidth, &user.DownloadBandwidth, &user.ExpirationDate, &user.LastLogin, &user.Status, &filters, &fsConfig,
&additionalInfo, &description, &email, &user.CreatedAt, &user.UpdatedAt, &user.UploadDataTransfer, &user.DownloadDataTransfer,
&user.TotalDataTransfer, &user.UsedUploadDataTransfer, &user.UsedDownloadDataTransfer, &user.DeletedAt, &user.FirstDownload,
&user.FirstUpload, &role, &user.LastPasswordChange)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return user, util.NewRecordNotFoundError(err.Error())
}
return user, err
}
if password.Valid {
user.Password = password.String
}
perms := make(map[string][]string)
err = json.Unmarshal(permissions, &perms)
if err != nil {
providerLog(logger.LevelError, "unable to deserialize permissions for user %q: %v", user.Username, err)
return user, fmt.Errorf("unable to deserialize permissions for user %q: %v", user.Username, err)
}
user.Permissions = perms
// we can have a empty string or an invalid json in null string
// so we do a relaxed test if the field is optional, for example we
// populate public keys only if unmarshal does not return an error
var pKeys []string
err = json.Unmarshal(publicKey, &pKeys)
if err == nil {
user.PublicKeys = pKeys
}
var userFilters UserFilters
err = json.Unmarshal(filters, &userFilters)
if err == nil {
user.Filters = userFilters
}
var fs vfs.Filesystem
err = json.Unmarshal(fsConfig, &fs)
if err == nil {
user.FsConfig = fs
}
if additionalInfo.Valid {
user.AdditionalInfo = additionalInfo.String
}
if description.Valid {
user.Description = description.String
}
if email.Valid {
user.Email = email.String
}
if role.Valid {
user.Role = role.String
}
user.SetEmptySecretsIfNil()
return user, nil
}
func sqlCommonGetFolder(ctx context.Context, name string, dbHandle sqlQuerier) (vfs.BaseVirtualFolder, error) {
var folder vfs.BaseVirtualFolder
q := getFolderByNameQuery()
row := dbHandle.QueryRowContext(ctx, q, name)
var mappedPath, description sql.NullString
var fsConfig []byte
err := row.Scan(&folder.ID, &mappedPath, &folder.UsedQuotaSize, &folder.UsedQuotaFiles, &folder.LastQuotaUpdate,
&folder.Name, &description, &fsConfig)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return folder, util.NewRecordNotFoundError(err.Error())
}
return folder, err
}
if mappedPath.Valid {
folder.MappedPath = mappedPath.String
}
if description.Valid {
folder.Description = description.String
}
var fs vfs.Filesystem
err = json.Unmarshal(fsConfig, &fs)
if err == nil {
folder.FsConfig = fs
}
return folder, err
}
func sqlCommonGetFolderByName(ctx context.Context, name string, dbHandle sqlQuerier) (vfs.BaseVirtualFolder, error) {
folder, err := sqlCommonGetFolder(ctx, name, dbHandle)
if err != nil {
return folder, err
}
folders, err := getVirtualFoldersWithUsers([]vfs.BaseVirtualFolder{folder}, dbHandle)
if err != nil {
return folder, err
}
if len(folders) != 1 {
return folder, fmt.Errorf("unable to associate users with folder %q", name)
}
folders, err = getVirtualFoldersWithGroups([]vfs.BaseVirtualFolder{folders[0]}, dbHandle)
if err != nil {
return folder, err
}
if len(folders) != 1 {
return folder, fmt.Errorf("unable to associate groups with folder %q", name)
}
return folders[0], nil
}
func sqlCommonAddFolder(folder *vfs.BaseVirtualFolder, dbHandle sqlQuerier) error {
err := ValidateFolder(folder)
if err != nil {
return err
}
fsConfig, err := json.Marshal(folder.FsConfig)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAddFolderQuery()
_, err = dbHandle.ExecContext(ctx, q, folder.MappedPath, folder.UsedQuotaSize, folder.UsedQuotaFiles,
folder.LastQuotaUpdate, folder.Name, folder.Description, fsConfig)
return err
}
func sqlCommonUpdateFolder(folder *vfs.BaseVirtualFolder, dbHandle sqlQuerier) error {
err := ValidateFolder(folder)
if err != nil {
return err
}
fsConfig, err := json.Marshal(folder.FsConfig)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateFolderQuery()
res, err := dbHandle.ExecContext(ctx, q, folder.MappedPath, folder.Description, fsConfig, folder.Name)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonDeleteFolder(folder vfs.BaseVirtualFolder, dbHandle sqlQuerier) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDeleteFolderQuery()
res, err := dbHandle.ExecContext(ctx, q, folder.Name)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonDumpFolders(dbHandle sqlQuerier) ([]vfs.BaseVirtualFolder, error) {
folders := make([]vfs.BaseVirtualFolder, 0, 50)
ctx, cancel := context.WithTimeout(context.Background(), longSQLQueryTimeout)
defer cancel()
q := getDumpFoldersQuery()
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return folders, err
}
defer rows.Close()
for rows.Next() {
var folder vfs.BaseVirtualFolder
var mappedPath, description sql.NullString
var fsConfig []byte
err = rows.Scan(&folder.ID, &mappedPath, &folder.UsedQuotaSize, &folder.UsedQuotaFiles,
&folder.LastQuotaUpdate, &folder.Name, &description, &fsConfig)
if err != nil {
return folders, err
}
if mappedPath.Valid {
folder.MappedPath = mappedPath.String
}
if description.Valid {
folder.Description = description.String
}
var fs vfs.Filesystem
err = json.Unmarshal(fsConfig, &fs)
if err == nil {
folder.FsConfig = fs
}
folders = append(folders, folder)
}
return folders, rows.Err()
}
func sqlCommonGetFolders(limit, offset int, order string, minimal bool, dbHandle sqlQuerier) ([]vfs.BaseVirtualFolder, error) {
folders := make([]vfs.BaseVirtualFolder, 0, limit)
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getFoldersQuery(order, minimal)
rows, err := dbHandle.QueryContext(ctx, q, limit, offset)
if err != nil {
return folders, err
}
defer rows.Close()
for rows.Next() {
var folder vfs.BaseVirtualFolder
if minimal {
err = rows.Scan(&folder.ID, &folder.Name)
if err != nil {
return folders, err
}
} else {
var mappedPath, description sql.NullString
var fsConfig []byte
err = rows.Scan(&folder.ID, &mappedPath, &folder.UsedQuotaSize, &folder.UsedQuotaFiles,
&folder.LastQuotaUpdate, &folder.Name, &description, &fsConfig)
if err != nil {
return folders, err
}
if mappedPath.Valid {
folder.MappedPath = mappedPath.String
}
if description.Valid {
folder.Description = description.String
}
var fs vfs.Filesystem
err = json.Unmarshal(fsConfig, &fs)
if err == nil {
folder.FsConfig = fs
}
}
folder.PrepareForRendering()
folders = append(folders, folder)
}
err = rows.Err()
if err != nil {
return folders, err
}
if minimal {
return folders, nil
}
folders, err = getVirtualFoldersWithUsers(folders, dbHandle)
if err != nil {
return folders, err
}
return getVirtualFoldersWithGroups(folders, dbHandle)
}
func sqlCommonClearUserFolderMapping(ctx context.Context, user *User, dbHandle sqlQuerier) error {
q := getClearUserFolderMappingQuery()
_, err := dbHandle.ExecContext(ctx, q, user.Username)
return err
}
func sqlCommonClearGroupFolderMapping(ctx context.Context, group *Group, dbHandle sqlQuerier) error {
q := getClearGroupFolderMappingQuery()
_, err := dbHandle.ExecContext(ctx, q, group.Name)
return err
}
func sqlCommonClearUserGroupMapping(ctx context.Context, user *User, dbHandle sqlQuerier) error {
q := getClearUserGroupMappingQuery()
_, err := dbHandle.ExecContext(ctx, q, user.Username)
return err
}
func sqlCommonAddUserFolderMapping(ctx context.Context, user *User, folder *vfs.VirtualFolder, dbHandle sqlQuerier) error {
q := getAddUserFolderMappingQuery()
_, err := dbHandle.ExecContext(ctx, q, folder.VirtualPath, folder.QuotaSize, folder.QuotaFiles, folder.Name, user.Username)
return err
}
func sqlCommonClearAdminGroupMapping(ctx context.Context, admin *Admin, dbHandle sqlQuerier) error {
q := getClearAdminGroupMappingQuery()
_, err := dbHandle.ExecContext(ctx, q, admin.Username)
return err
}
func sqlCommonAddGroupFolderMapping(ctx context.Context, group *Group, folder *vfs.VirtualFolder, dbHandle sqlQuerier) error {
q := getAddGroupFolderMappingQuery()
_, err := dbHandle.ExecContext(ctx, q, folder.VirtualPath, folder.QuotaSize, folder.QuotaFiles, folder.Name, group.Name)
return err
}
func sqlCommonAddUserGroupMapping(ctx context.Context, username, groupName string, groupType int, dbHandle sqlQuerier) error {
q := getAddUserGroupMappingQuery()
_, err := dbHandle.ExecContext(ctx, q, username, groupName, groupType)
return err
}
func sqlCommonAddAdminGroupMapping(ctx context.Context, username, groupName string, mappingOptions AdminGroupMappingOptions,
dbHandle sqlQuerier,
) error {
options, err := json.Marshal(mappingOptions)
if err != nil {
return err
}
q := getAddAdminGroupMappingQuery()
_, err = dbHandle.ExecContext(ctx, q, username, groupName, options)
return err
}
func generateGroupVirtualFoldersMapping(ctx context.Context, group *Group, dbHandle sqlQuerier) error {
err := sqlCommonClearGroupFolderMapping(ctx, group, dbHandle)
if err != nil {
return err
}
for idx := range group.VirtualFolders {
vfolder := &group.VirtualFolders[idx]
err = sqlCommonAddGroupFolderMapping(ctx, group, vfolder, dbHandle)
if err != nil {
return err
}
}
return err
}
func generateUserVirtualFoldersMapping(ctx context.Context, user *User, dbHandle sqlQuerier) error {
err := sqlCommonClearUserFolderMapping(ctx, user, dbHandle)
if err != nil {
return err
}
for idx := range user.VirtualFolders {
vfolder := &user.VirtualFolders[idx]
err = sqlCommonAddUserFolderMapping(ctx, user, vfolder, dbHandle)
if err != nil {
return err
}
}
return err
}
func generateUserGroupMapping(ctx context.Context, user *User, dbHandle sqlQuerier) error {
err := sqlCommonClearUserGroupMapping(ctx, user, dbHandle)
if err != nil {
return err
}
for _, group := range user.Groups {
err = sqlCommonAddUserGroupMapping(ctx, user.Username, group.Name, group.Type, dbHandle)
if err != nil {
return err
}
}
return err
}
func generateAdminGroupMapping(ctx context.Context, admin *Admin, dbHandle sqlQuerier) error {
err := sqlCommonClearAdminGroupMapping(ctx, admin, dbHandle)
if err != nil {
return err
}
for _, group := range admin.Groups {
err = sqlCommonAddAdminGroupMapping(ctx, admin.Username, group.Name, group.Options, dbHandle)
if err != nil {
return err
}
}
return err
}
func getDefenderHostsWithScores(ctx context.Context, hosts []DefenderEntry, from int64, idForScores []int64,
dbHandle sqlQuerier) (
[]DefenderEntry,
error,
) {
if len(idForScores) == 0 {
return hosts, nil
}
hostsWithScores := make(map[int64]int)
q := getDefenderEventsQuery(idForScores)
rows, err := dbHandle.QueryContext(ctx, q, from)
if err != nil {
providerLog(logger.LevelError, "unable to get score for hosts with id %+v: %v", idForScores, err)
return nil, err
}
defer rows.Close()
for rows.Next() {
var hostID int64
var score int
err = rows.Scan(&hostID, &score)
if err != nil {
providerLog(logger.LevelError, "error scanning host score row: %v", err)
return hosts, err
}
if score > 0 {
hostsWithScores[hostID] = score
}
}
err = rows.Err()
if err != nil {
return hosts, err
}
result := make([]DefenderEntry, 0, len(hosts))
for idx := range hosts {
hosts[idx].Score = hostsWithScores[hosts[idx].ID]
if hosts[idx].Score > 0 || !hosts[idx].BanTime.IsZero() {
result = append(result, hosts[idx])
}
}
return result, nil
}
func getAdminWithGroups(ctx context.Context, admin Admin, dbHandle sqlQuerier) (Admin, error) {
admins, err := getAdminsWithGroups(ctx, []Admin{admin}, dbHandle)
if err != nil {
return admin, err
}
if len(admins) == 0 {
return admin, errSQLGroupsAssociation
}
return admins[0], err
}
func getAdminsWithGroups(ctx context.Context, admins []Admin, dbHandle sqlQuerier) ([]Admin, error) {
if len(admins) == 0 {
return admins, nil
}
adminsGroups := make(map[int64][]AdminGroupMapping)
q := getRelatedGroupsForAdminsQuery(admins)
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var group AdminGroupMapping
var adminID int64
var options []byte
err = rows.Scan(&group.Name, &options, &adminID)
if err != nil {
return admins, err
}
err = json.Unmarshal(options, &group.Options)
if err != nil {
return admins, err
}
adminsGroups[adminID] = append(adminsGroups[adminID], group)
}
err = rows.Err()
if err != nil {
return admins, err
}
if len(adminsGroups) == 0 {
return admins, err
}
for idx := range admins {
ref := &admins[idx]
ref.Groups = adminsGroups[ref.ID]
}
return admins, err
}
func getUserWithVirtualFolders(ctx context.Context, user User, dbHandle sqlQuerier) (User, error) {
users, err := getUsersWithVirtualFolders(ctx, []User{user}, dbHandle)
if err != nil {
return user, err
}
if len(users) == 0 {
return user, errSQLFoldersAssociation
}
return users[0], err
}
func getUsersWithVirtualFolders(ctx context.Context, users []User, dbHandle sqlQuerier) ([]User, error) {
if len(users) == 0 {
return users, nil
}
usersVirtualFolders := make(map[int64][]vfs.VirtualFolder)
q := getRelatedFoldersForUsersQuery(users)
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var folder vfs.VirtualFolder
var userID int64
var mappedPath, description sql.NullString
var fsConfig []byte
err = rows.Scan(&folder.ID, &folder.Name, &mappedPath, &folder.UsedQuotaSize, &folder.UsedQuotaFiles,
&folder.LastQuotaUpdate, &folder.VirtualPath, &folder.QuotaSize, &folder.QuotaFiles, &userID, &fsConfig,
&description)
if err != nil {
return users, err
}
if mappedPath.Valid {
folder.MappedPath = mappedPath.String
}
if description.Valid {
folder.Description = description.String
}
var fs vfs.Filesystem
err = json.Unmarshal(fsConfig, &fs)
if err == nil {
folder.FsConfig = fs
}
usersVirtualFolders[userID] = append(usersVirtualFolders[userID], folder)
}
err = rows.Err()
if err != nil {
return users, err
}
if len(usersVirtualFolders) == 0 {
return users, err
}
for idx := range users {
ref := &users[idx]
ref.VirtualFolders = usersVirtualFolders[ref.ID]
}
return users, err
}
func getUserWithGroups(ctx context.Context, user User, dbHandle sqlQuerier) (User, error) {
users, err := getUsersWithGroups(ctx, []User{user}, dbHandle)
if err != nil {
return user, err
}
if len(users) == 0 {
return user, errSQLGroupsAssociation
}
return users[0], err
}
func getUsersWithGroups(ctx context.Context, users []User, dbHandle sqlQuerier) ([]User, error) {
if len(users) == 0 {
return users, nil
}
usersGroups := make(map[int64][]sdk.GroupMapping)
q := getRelatedGroupsForUsersQuery(users)
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var group sdk.GroupMapping
var userID int64
err = rows.Scan(&group.Name, &group.Type, &userID)
if err != nil {
return users, err
}
usersGroups[userID] = append(usersGroups[userID], group)
}
err = rows.Err()
if err != nil {
return users, err
}
if len(usersGroups) == 0 {
return users, err
}
for idx := range users {
ref := &users[idx]
ref.Groups = usersGroups[ref.ID]
}
return users, err
}
func getGroupWithUsers(ctx context.Context, group Group, dbHandle sqlQuerier) (Group, error) {
groups, err := getGroupsWithUsers(ctx, []Group{group}, dbHandle)
if err != nil {
return group, err
}
if len(groups) == 0 {
return group, errSQLUsersAssociation
}
return groups[0], err
}
func getRoleWithUsers(ctx context.Context, role Role, dbHandle sqlQuerier) (Role, error) {
roles, err := getRolesWithUsers(ctx, []Role{role}, dbHandle)
if err != nil {
return role, err
}
if len(roles) == 0 {
return role, errors.New("unable to associate users with role")
}
return roles[0], err
}
func getRoleWithAdmins(ctx context.Context, role Role, dbHandle sqlQuerier) (Role, error) {
roles, err := getRolesWithAdmins(ctx, []Role{role}, dbHandle)
if err != nil {
return role, err
}
if len(roles) == 0 {
return role, errors.New("unable to associate admins with role")
}
return roles[0], err
}
func getGroupWithAdmins(ctx context.Context, group Group, dbHandle sqlQuerier) (Group, error) {
groups, err := getGroupsWithAdmins(ctx, []Group{group}, dbHandle)
if err != nil {
return group, err
}
if len(groups) == 0 {
return group, errSQLUsersAssociation
}
return groups[0], err
}
func getGroupWithVirtualFolders(ctx context.Context, group Group, dbHandle sqlQuerier) (Group, error) {
groups, err := getGroupsWithVirtualFolders(ctx, []Group{group}, dbHandle)
if err != nil {
return group, err
}
if len(groups) == 0 {
return group, errSQLFoldersAssociation
}
return groups[0], err
}
func getGroupsWithVirtualFolders(ctx context.Context, groups []Group, dbHandle sqlQuerier) ([]Group, error) {
if len(groups) == 0 {
return groups, nil
}
q := getRelatedFoldersForGroupsQuery(groups)
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
groupsVirtualFolders := make(map[int64][]vfs.VirtualFolder)
for rows.Next() {
var groupID int64
var folder vfs.VirtualFolder
var mappedPath, description sql.NullString
var fsConfig []byte
err = rows.Scan(&folder.ID, &folder.Name, &mappedPath, &folder.UsedQuotaSize, &folder.UsedQuotaFiles,
&folder.LastQuotaUpdate, &folder.VirtualPath, &folder.QuotaSize, &folder.QuotaFiles, &groupID, &fsConfig,
&description)
if err != nil {
return groups, err
}
if mappedPath.Valid {
folder.MappedPath = mappedPath.String
}
if description.Valid {
folder.Description = description.String
}
var fs vfs.Filesystem
err = json.Unmarshal(fsConfig, &fs)
if err == nil {
folder.FsConfig = fs
}
groupsVirtualFolders[groupID] = append(groupsVirtualFolders[groupID], folder)
}
err = rows.Err()
if err != nil {
return groups, err
}
if len(groupsVirtualFolders) == 0 {
return groups, err
}
for idx := range groups {
ref := &groups[idx]
ref.VirtualFolders = groupsVirtualFolders[ref.ID]
}
return groups, err
}
func getGroupsWithUsers(ctx context.Context, groups []Group, dbHandle sqlQuerier) ([]Group, error) {
if len(groups) == 0 {
return groups, nil
}
q := getRelatedUsersForGroupsQuery(groups)
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
groupsUsers := make(map[int64][]string)
for rows.Next() {
var username string
var groupID int64
err = rows.Scan(&groupID, &username)
if err != nil {
return groups, err
}
groupsUsers[groupID] = append(groupsUsers[groupID], username)
}
err = rows.Err()
if err != nil {
return groups, err
}
if len(groupsUsers) == 0 {
return groups, err
}
for idx := range groups {
ref := &groups[idx]
ref.Users = groupsUsers[ref.ID]
}
return groups, err
}
func getRolesWithUsers(ctx context.Context, roles []Role, dbHandle sqlQuerier) ([]Role, error) {
if len(roles) == 0 {
return roles, nil
}
rows, err := dbHandle.QueryContext(ctx, getUsersWithRolesQuery(roles))
if err != nil {
return nil, err
}
defer rows.Close()
rolesUsers := make(map[int64][]string)
for rows.Next() {
var roleID int64
var username string
err = rows.Scan(&roleID, &username)
if err != nil {
return roles, err
}
rolesUsers[roleID] = append(rolesUsers[roleID], username)
}
err = rows.Err()
if err != nil {
return roles, err
}
if len(rolesUsers) > 0 {
for idx := range roles {
ref := &roles[idx]
ref.Users = rolesUsers[ref.ID]
}
}
return roles, nil
}
func getRolesWithAdmins(ctx context.Context, roles []Role, dbHandle sqlQuerier) ([]Role, error) {
if len(roles) == 0 {
return roles, nil
}
rows, err := dbHandle.QueryContext(ctx, getAdminsWithRolesQuery(roles))
if err != nil {
return nil, err
}
defer rows.Close()
rolesAdmins := make(map[int64][]string)
for rows.Next() {
var roleID int64
var username string
err = rows.Scan(&roleID, &username)
if err != nil {
return roles, err
}
rolesAdmins[roleID] = append(rolesAdmins[roleID], username)
}
if err = rows.Err(); err != nil {
return roles, err
}
if len(rolesAdmins) > 0 {
for idx := range roles {
ref := &roles[idx]
ref.Admins = rolesAdmins[ref.ID]
}
}
return roles, nil
}
func getGroupsWithAdmins(ctx context.Context, groups []Group, dbHandle sqlQuerier) ([]Group, error) {
if len(groups) == 0 {
return groups, nil
}
q := getRelatedAdminsForGroupsQuery(groups)
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
groupsAdmins := make(map[int64][]string)
for rows.Next() {
var groupID int64
var username string
err = rows.Scan(&groupID, &username)
if err != nil {
return groups, err
}
groupsAdmins[groupID] = append(groupsAdmins[groupID], username)
}
err = rows.Err()
if err != nil {
return groups, err
}
if len(groupsAdmins) > 0 {
for idx := range groups {
ref := &groups[idx]
ref.Admins = groupsAdmins[ref.ID]
}
}
return groups, nil
}
func getVirtualFoldersWithGroups(folders []vfs.BaseVirtualFolder, dbHandle sqlQuerier) ([]vfs.BaseVirtualFolder, error) {
if len(folders) == 0 {
return folders, nil
}
vFoldersGroups := make(map[int64][]string)
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getRelatedGroupsForFoldersQuery(folders)
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var name string
var folderID int64
err = rows.Scan(&folderID, &name)
if err != nil {
return folders, err
}
vFoldersGroups[folderID] = append(vFoldersGroups[folderID], name)
}
err = rows.Err()
if err != nil {
return folders, err
}
if len(vFoldersGroups) == 0 {
return folders, err
}
for idx := range folders {
ref := &folders[idx]
ref.Groups = vFoldersGroups[ref.ID]
}
return folders, err
}
func getVirtualFoldersWithUsers(folders []vfs.BaseVirtualFolder, dbHandle sqlQuerier) ([]vfs.BaseVirtualFolder, error) {
if len(folders) == 0 {
return folders, nil
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getRelatedUsersForFoldersQuery(folders)
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
vFoldersUsers := make(map[int64][]string)
for rows.Next() {
var username string
var folderID int64
err = rows.Scan(&folderID, &username)
if err != nil {
return folders, err
}
vFoldersUsers[folderID] = append(vFoldersUsers[folderID], username)
}
err = rows.Err()
if err != nil {
return folders, err
}
if len(vFoldersUsers) == 0 {
return folders, err
}
for idx := range folders {
ref := &folders[idx]
ref.Users = vFoldersUsers[ref.ID]
}
return folders, err
}
func sqlCommonUpdateFolderQuota(name string, filesAdd int, sizeAdd int64, reset bool, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateFolderQuotaQuery(reset)
_, err := dbHandle.ExecContext(ctx, q, sizeAdd, filesAdd, util.GetTimeAsMsSinceEpoch(time.Now()), name)
if err == nil {
providerLog(logger.LevelDebug, "quota updated for folder %q, files increment: %d size increment: %d is reset? %t",
name, filesAdd, sizeAdd, reset)
} else {
providerLog(logger.LevelWarn, "error updating quota for folder %q: %v", name, err)
}
return err
}
func sqlCommonGetFolderUsedQuota(mappedPath string, dbHandle *sql.DB) (int, int64, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getQuotaFolderQuery()
var usedFiles int
var usedSize int64
err := dbHandle.QueryRowContext(ctx, q, mappedPath).Scan(&usedSize, &usedFiles)
if err != nil {
providerLog(logger.LevelError, "error getting quota for folder: %v, error: %v", mappedPath, err)
return 0, 0, err
}
return usedFiles, usedSize, err
}
func getAPIKeyWithRelatedFields(ctx context.Context, apiKey APIKey, dbHandle sqlQuerier) (APIKey, error) {
var apiKeys []APIKey
var err error
scope := APIKeyScopeAdmin
if apiKey.userID > 0 {
scope = APIKeyScopeUser
}
apiKeys, err = getRelatedValuesForAPIKeys(ctx, []APIKey{apiKey}, dbHandle, scope)
if err != nil {
return apiKey, err
}
if len(apiKeys) > 0 {
apiKey = apiKeys[0]
}
return apiKey, nil
}
func getRelatedValuesForAPIKeys(ctx context.Context, apiKeys []APIKey, dbHandle sqlQuerier, scope APIKeyScope) ([]APIKey, error) {
if len(apiKeys) == 0 {
return apiKeys, nil
}
values := make(map[int64]string)
var q string
if scope == APIKeyScopeUser {
q = getRelatedUsersForAPIKeysQuery(apiKeys)
} else {
q = getRelatedAdminsForAPIKeysQuery(apiKeys)
}
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var valueID int64
var valueName string
err = rows.Scan(&valueID, &valueName)
if err != nil {
return apiKeys, err
}
values[valueID] = valueName
}
err = rows.Err()
if err != nil {
return apiKeys, err
}
if len(values) == 0 {
return apiKeys, nil
}
for idx := range apiKeys {
ref := &apiKeys[idx]
if scope == APIKeyScopeUser {
ref.User = values[ref.userID]
} else {
ref.Admin = values[ref.adminID]
}
}
return apiKeys, nil
}
func sqlCommonGetAPIKeyRelatedIDs(apiKey *APIKey) (sql.NullInt64, sql.NullInt64, error) {
var userID, adminID sql.NullInt64
if apiKey.User != "" {
u, err := provider.userExists(apiKey.User, "")
if err != nil {
return userID, adminID, util.NewGenericError(fmt.Sprintf("unable to validate user %v", apiKey.User))
}
userID.Valid = true
userID.Int64 = u.ID
}
if apiKey.Admin != "" {
a, err := provider.adminExists(apiKey.Admin)
if err != nil {
return userID, adminID, util.NewValidationError(fmt.Sprintf("unable to validate admin %v", apiKey.Admin))
}
adminID.Valid = true
adminID.Int64 = a.ID
}
return userID, adminID, nil
}
func sqlCommonAddSession(session Session, dbHandle *sql.DB) error {
if err := session.validate(); err != nil {
return err
}
data, err := json.Marshal(session.Data)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAddSessionQuery()
_, err = dbHandle.ExecContext(ctx, q, session.Key, data, session.Type, session.Timestamp)
return err
}
func sqlCommonGetSession(key string, dbHandle sqlQuerier) (Session, error) {
var session Session
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getSessionQuery()
var data []byte // type hint, some driver will use string instead of []byte if the type is any
err := dbHandle.QueryRowContext(ctx, q, key).Scan(&session.Key, &data, &session.Type, &session.Timestamp)
if err != nil {
return session, err
}
session.Data = data
return session, nil
}
func sqlCommonDeleteSession(key string, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDeleteSessionQuery()
res, err := dbHandle.ExecContext(ctx, q, key)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonCleanupSessions(sessionType SessionType, before int64, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getCleanupSessionsQuery()
_, err := dbHandle.ExecContext(ctx, q, sessionType, before)
return err
}
func getActionsWithRuleNames(ctx context.Context, actions []BaseEventAction, dbHandle sqlQuerier,
) ([]BaseEventAction, error) {
if len(actions) == 0 {
return actions, nil
}
q := getRelatedRulesForActionsQuery(actions)
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
actionsRules := make(map[int64][]string)
for rows.Next() {
var name string
var actionID int64
if err = rows.Scan(&actionID, &name); err != nil {
return nil, err
}
actionsRules[actionID] = append(actionsRules[actionID], name)
}
err = rows.Err()
if err != nil {
return nil, err
}
if len(actionsRules) == 0 {
return actions, nil
}
for idx := range actions {
ref := &actions[idx]
ref.Rules = actionsRules[ref.ID]
}
return actions, nil
}
func getRulesWithActions(ctx context.Context, rules []EventRule, dbHandle sqlQuerier) ([]EventRule, error) {
if len(rules) == 0 {
return rules, nil
}
rulesActions := make(map[int64][]EventAction)
q := getRelatedActionsForRulesQuery(rules)
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return nil, err
}
defer rows.Close()
for rows.Next() {
var action EventAction
var ruleID int64
var description sql.NullString
var baseOptions, options []byte
err = rows.Scan(&action.ID, &action.Name, &description, &action.Type, &baseOptions, &options,
&action.Order, &ruleID)
if err != nil {
return rules, err
}
if len(baseOptions) > 0 {
err = json.Unmarshal(baseOptions, &action.BaseEventAction.Options)
if err != nil {
return rules, err
}
}
if len(options) > 0 {
err = json.Unmarshal(options, &action.Options)
if err != nil {
return rules, err
}
}
action.BaseEventAction.Options.SetEmptySecretsIfNil()
rulesActions[ruleID] = append(rulesActions[ruleID], action)
}
err = rows.Err()
if err != nil {
return rules, err
}
if len(rulesActions) == 0 {
return rules, nil
}
for idx := range rules {
ref := &rules[idx]
ref.Actions = rulesActions[ref.ID]
}
return rules, nil
}
func generateEventRuleActionsMapping(ctx context.Context, rule *EventRule, dbHandle sqlQuerier) error {
q := getClearRuleActionMappingQuery()
_, err := dbHandle.ExecContext(ctx, q, rule.Name)
if err != nil {
return err
}
for _, action := range rule.Actions {
options, err := json.Marshal(action.Options)
if err != nil {
return err
}
q = getAddRuleActionMappingQuery()
_, err = dbHandle.ExecContext(ctx, q, rule.Name, action.Name, action.Order, options)
if err != nil {
return err
}
}
return nil
}
func sqlCommonGetEventActionByName(name string, dbHandle sqlQuerier) (BaseEventAction, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getEventActionByNameQuery()
row := dbHandle.QueryRowContext(ctx, q, name)
action, err := getEventActionFromDbRow(row)
if err != nil {
return action, err
}
actions, err := getActionsWithRuleNames(ctx, []BaseEventAction{action}, dbHandle)
if err != nil {
return action, err
}
if len(actions) != 1 {
return action, fmt.Errorf("unable to associate rules with action %q", name)
}
return actions[0], nil
}
func sqlCommonDumpEventActions(dbHandle sqlQuerier) ([]BaseEventAction, error) {
actions := make([]BaseEventAction, 0, 10)
ctx, cancel := context.WithTimeout(context.Background(), longSQLQueryTimeout)
defer cancel()
q := getDumpEventActionsQuery()
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return actions, err
}
defer rows.Close()
for rows.Next() {
action, err := getEventActionFromDbRow(rows)
if err != nil {
return actions, err
}
actions = append(actions, action)
}
return actions, rows.Err()
}
func sqlCommonGetEventActions(limit int, offset int, order string, minimal bool,
dbHandle sqlQuerier,
) ([]BaseEventAction, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getEventsActionsQuery(order, minimal)
actions := make([]BaseEventAction, 0, limit)
rows, err := dbHandle.QueryContext(ctx, q, limit, offset)
if err != nil {
return actions, err
}
defer rows.Close()
for rows.Next() {
var action BaseEventAction
if minimal {
err = rows.Scan(&action.ID, &action.Name)
} else {
action, err = getEventActionFromDbRow(rows)
}
if err != nil {
return actions, err
}
actions = append(actions, action)
}
err = rows.Err()
if err != nil {
return nil, err
}
if minimal {
return actions, nil
}
actions, err = getActionsWithRuleNames(ctx, actions, dbHandle)
if err != nil {
return nil, err
}
for idx := range actions {
actions[idx].PrepareForRendering()
}
return actions, nil
}
func sqlCommonAddEventAction(action *BaseEventAction, dbHandle *sql.DB) error {
if err := action.validate(); err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAddEventActionQuery()
options, err := json.Marshal(action.Options)
if err != nil {
return err
}
_, err = dbHandle.ExecContext(ctx, q, action.Name, action.Description, action.Type, options)
return err
}
func sqlCommonUpdateEventAction(action *BaseEventAction, dbHandle *sql.DB) error {
if err := action.validate(); err != nil {
return err
}
options, err := json.Marshal(action.Options)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
q := getUpdateEventActionQuery()
res, err := tx.ExecContext(ctx, q, action.Description, action.Type, options, action.Name)
if err != nil {
return err
}
if err := sqlCommonRequireRowAffected(res); err != nil {
return err
}
q = getUpdateRulesTimestampQuery()
_, err = tx.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), action.Name)
return err
})
}
func sqlCommonDeleteEventAction(action BaseEventAction, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDeleteEventActionQuery()
res, err := dbHandle.ExecContext(ctx, q, action.Name)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonGetEventRuleByName(name string, dbHandle sqlQuerier) (EventRule, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getEventRulesByNameQuery()
row := dbHandle.QueryRowContext(ctx, q, name)
rule, err := getEventRuleFromDbRow(row)
if err != nil {
return rule, err
}
rules, err := getRulesWithActions(ctx, []EventRule{rule}, dbHandle)
if err != nil {
return rule, err
}
if len(rules) != 1 {
return rule, fmt.Errorf("unable to associate rule %q with actions", name)
}
return rules[0], nil
}
func sqlCommonDumpEventRules(dbHandle sqlQuerier) ([]EventRule, error) {
rules := make([]EventRule, 0, 10)
ctx, cancel := context.WithTimeout(context.Background(), longSQLQueryTimeout)
defer cancel()
q := getDumpEventRulesQuery()
rows, err := dbHandle.QueryContext(ctx, q)
if err != nil {
return rules, err
}
defer rows.Close()
for rows.Next() {
rule, err := getEventRuleFromDbRow(rows)
if err != nil {
return rules, err
}
rules = append(rules, rule)
}
err = rows.Err()
if err != nil {
return rules, err
}
return getRulesWithActions(ctx, rules, dbHandle)
}
func sqlCommonGetRecentlyUpdatedRules(after int64, dbHandle sqlQuerier) ([]EventRule, error) {
rules := make([]EventRule, 0, 10)
ctx, cancel := context.WithTimeout(context.Background(), longSQLQueryTimeout)
defer cancel()
q := getRecentlyUpdatedRulesQuery()
rows, err := dbHandle.QueryContext(ctx, q, after)
if err != nil {
return rules, err
}
defer rows.Close()
for rows.Next() {
rule, err := getEventRuleFromDbRow(rows)
if err != nil {
return rules, err
}
rules = append(rules, rule)
}
err = rows.Err()
if err != nil {
return rules, err
}
return getRulesWithActions(ctx, rules, dbHandle)
}
func sqlCommonGetEventRules(limit int, offset int, order string, dbHandle sqlQuerier) ([]EventRule, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getEventRulesQuery(order)
rules := make([]EventRule, 0, limit)
rows, err := dbHandle.QueryContext(ctx, q, limit, offset)
if err != nil {
return rules, err
}
defer rows.Close()
for rows.Next() {
rule, err := getEventRuleFromDbRow(rows)
if err != nil {
return rules, err
}
rules = append(rules, rule)
}
err = rows.Err()
if err != nil {
return rules, err
}
rules, err = getRulesWithActions(ctx, rules, dbHandle)
if err != nil {
return rules, err
}
for idx := range rules {
rules[idx].PrepareForRendering()
}
return rules, nil
}
func sqlCommonAddEventRule(rule *EventRule, dbHandle *sql.DB) error {
if err := rule.validate(); err != nil {
return err
}
conditions, err := json.Marshal(rule.Conditions)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
if config.IsShared == 1 {
_, err := tx.ExecContext(ctx, getRemoveSoftDeletedRuleQuery(), rule.Name)
if err != nil {
return err
}
}
q := getAddEventRuleQuery()
_, err := tx.ExecContext(ctx, q, rule.Name, rule.Description, util.GetTimeAsMsSinceEpoch(time.Now()),
util.GetTimeAsMsSinceEpoch(time.Now()), rule.Trigger, conditions, rule.Status)
if err != nil {
return err
}
return generateEventRuleActionsMapping(ctx, rule, tx)
})
}
func sqlCommonUpdateEventRule(rule *EventRule, dbHandle *sql.DB) error {
if err := rule.validate(); err != nil {
return err
}
conditions, err := json.Marshal(rule.Conditions)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
q := getUpdateEventRuleQuery()
_, err := tx.ExecContext(ctx, q, rule.Description, util.GetTimeAsMsSinceEpoch(time.Now()),
rule.Trigger, conditions, rule.Status, rule.Name)
if err != nil {
return err
}
return generateEventRuleActionsMapping(ctx, rule, tx)
})
}
func sqlCommonDeleteEventRule(rule EventRule, softDelete bool, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
if softDelete {
q := getClearRuleActionMappingQuery()
_, err := tx.ExecContext(ctx, q, rule.Name)
if err != nil {
return err
}
}
q := getDeleteEventRuleQuery(softDelete)
if softDelete {
ts := util.GetTimeAsMsSinceEpoch(time.Now())
res, err := tx.ExecContext(ctx, q, ts, ts, rule.Name)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
res, err := tx.ExecContext(ctx, q, rule.Name)
if err != nil {
return err
}
if err = sqlCommonRequireRowAffected(res); err != nil {
return err
}
return sqlCommonDeleteTask(rule.Name, tx)
})
}
func sqlCommonGetTaskByName(name string, dbHandle sqlQuerier) (Task, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
task := Task{
Name: name,
}
q := getTaskByNameQuery()
row := dbHandle.QueryRowContext(ctx, q, name)
err := row.Scan(&task.UpdateAt, &task.Version)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return task, util.NewRecordNotFoundError(err.Error())
}
}
return task, err
}
func sqlCommonAddTask(name string, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAddTaskQuery()
_, err := dbHandle.ExecContext(ctx, q, name, util.GetTimeAsMsSinceEpoch(time.Now()))
return err
}
func sqlCommonUpdateTask(name string, version int64, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateTaskQuery()
res, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), name, version)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonUpdateTaskTimestamp(name string, dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateTaskTimestampQuery()
res, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), name)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonDeleteTask(name string, dbHandle sqlQuerier) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDeleteTaskQuery()
_, err := dbHandle.ExecContext(ctx, q, name)
return err
}
func sqlCommonAddNode(dbHandle *sql.DB) error {
if err := currentNode.validate(); err != nil {
return fmt.Errorf("unable to register cluster node: %w", err)
}
data, err := json.Marshal(currentNode.Data)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getAddNodeQuery()
_, err = dbHandle.ExecContext(ctx, q, currentNode.Name, data, util.GetTimeAsMsSinceEpoch(time.Now()),
util.GetTimeAsMsSinceEpoch(time.Now()))
if err != nil {
return fmt.Errorf("unable to register cluster node: %w", err)
}
providerLog(logger.LevelInfo, "registered as cluster node %q, port: %d, proto: %s",
currentNode.Name, currentNode.Data.Port, currentNode.Data.Proto)
return nil
}
func sqlCommonGetNodeByName(name string, dbHandle *sql.DB) (Node, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
var data []byte
var node Node
q := getNodeByNameQuery()
row := dbHandle.QueryRowContext(ctx, q, name, util.GetTimeAsMsSinceEpoch(time.Now().Add(activeNodeTimeDiff)))
err := row.Scan(&node.Name, &data, &node.CreatedAt, &node.UpdatedAt)
if err != nil {
if errors.Is(err, sql.ErrNoRows) {
return node, util.NewRecordNotFoundError(err.Error())
}
return node, err
}
err = json.Unmarshal(data, &node.Data)
return node, err
}
func sqlCommonGetNodes(dbHandle *sql.DB) ([]Node, error) {
var nodes []Node
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getNodesQuery()
rows, err := dbHandle.QueryContext(ctx, q, currentNode.Name,
util.GetTimeAsMsSinceEpoch(time.Now().Add(activeNodeTimeDiff)))
if err != nil {
return nodes, err
}
defer rows.Close()
for rows.Next() {
var node Node
var data []byte
err = rows.Scan(&node.Name, &data, &node.CreatedAt, &node.UpdatedAt)
if err != nil {
return nodes, err
}
err = json.Unmarshal(data, &node.Data)
if err != nil {
return nodes, err
}
nodes = append(nodes, node)
}
return nodes, rows.Err()
}
func sqlCommonUpdateNodeTimestamp(dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateNodeTimestampQuery()
res, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now()), currentNode.Name)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonCleanupNodes(dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getCleanupNodesQuery()
_, err := dbHandle.ExecContext(ctx, q, util.GetTimeAsMsSinceEpoch(time.Now().Add(10*activeNodeTimeDiff)))
return err
}
func sqlCommonGetConfigs(dbHandle sqlQuerier) (Configs, error) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
var result Configs
var configs []byte
q := getConfigsQuery()
err := dbHandle.QueryRowContext(ctx, q).Scan(&configs)
if err != nil {
return result, err
}
err = json.Unmarshal(configs, &result)
return result, err
}
func sqlCommonSetConfigs(configs *Configs, dbHandle *sql.DB) error {
if err := configs.validate(); err != nil {
return err
}
asJSON, err := json.Marshal(configs)
if err != nil {
return err
}
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getUpdateConfigsQuery()
res, err := dbHandle.ExecContext(ctx, q, asJSON)
if err != nil {
return err
}
return sqlCommonRequireRowAffected(res)
}
func sqlCommonGetDatabaseVersion(dbHandle sqlQuerier, showInitWarn bool) (schemaVersion, error) {
var result schemaVersion
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
q := getDatabaseVersionQuery()
stmt, err := dbHandle.PrepareContext(ctx, q)
if err != nil {
providerLog(logger.LevelError, "error preparing database query %q: %v", q, err)
if showInitWarn && strings.Contains(err.Error(), sqlTableSchemaVersion) {
logger.WarnToConsole("database query error, did you forgot to run the \"initprovider\" command?")
}
return result, err
}
defer stmt.Close()
row := stmt.QueryRowContext(ctx)
err = row.Scan(&result.Version)
return result, err
}
func sqlCommonRequireRowAffected(res sql.Result) error {
affected, err := res.RowsAffected()
if err == nil && affected == 0 {
return util.NewRecordNotFoundError(sql.ErrNoRows.Error())
}
return nil
}
func sqlCommonUpdateDatabaseVersion(ctx context.Context, dbHandle sqlQuerier, version int) error {
q := getUpdateDBVersionQuery()
_, err := dbHandle.ExecContext(ctx, q, version)
return err
}
func sqlCommonExecSQLAndUpdateDBVersion(dbHandle *sql.DB, sqlQueries []string, newVersion int, isUp bool) error {
if err := sqlAcquireLock(dbHandle); err != nil {
return err
}
defer sqlReleaseLock(dbHandle)
ctx, cancel := context.WithTimeout(context.Background(), longSQLQueryTimeout)
defer cancel()
if newVersion > 0 {
currentVersion, err := sqlCommonGetDatabaseVersion(dbHandle, false)
if err == nil {
if (isUp && currentVersion.Version >= newVersion) || (!isUp && currentVersion.Version <= newVersion) {
providerLog(logger.LevelInfo, "current schema version: %v, requested: %v, did you execute simultaneous migrations?",
currentVersion.Version, newVersion)
return nil
}
}
}
return sqlCommonExecuteTx(ctx, dbHandle, func(tx *sql.Tx) error {
for _, q := range sqlQueries {
if strings.TrimSpace(q) == "" {
continue
}
_, err := tx.ExecContext(ctx, q)
if err != nil {
return err
}
}
if newVersion == 0 {
return nil
}
return sqlCommonUpdateDatabaseVersion(ctx, tx, newVersion)
})
}
func sqlAcquireLock(dbHandle *sql.DB) error {
ctx, cancel := context.WithTimeout(context.Background(), longSQLQueryTimeout)
defer cancel()
switch config.Driver {
case PGSQLDataProviderName:
_, err := dbHandle.ExecContext(ctx, `SELECT pg_advisory_lock(101,1)`)
if err != nil {
return fmt.Errorf("unable to get advisory lock: %w", err)
}
providerLog(logger.LevelInfo, "acquired database lock")
case MySQLDataProviderName:
var lockResult sql.NullInt64
err := dbHandle.QueryRowContext(ctx, `SELECT GET_LOCK('sftpgo.migration',30)`).Scan(&lockResult)
if err != nil {
return fmt.Errorf("unable to get lock: %w", err)
}
if !lockResult.Valid {
return errors.New("unable to get lock: null value returned")
}
if lockResult.Int64 != 1 {
return fmt.Errorf("unable to get lock, result: %d", lockResult.Int64)
}
providerLog(logger.LevelInfo, "acquired database lock")
}
return nil
}
func sqlReleaseLock(dbHandle *sql.DB) {
ctx, cancel := context.WithTimeout(context.Background(), defaultSQLQueryTimeout)
defer cancel()
switch config.Driver {
case PGSQLDataProviderName:
_, err := dbHandle.ExecContext(ctx, `SELECT pg_advisory_unlock(101,1)`)
if err != nil {
providerLog(logger.LevelWarn, "unable to release lock: %v", err)
} else {
providerLog(logger.LevelInfo, "released database lock")
}
case MySQLDataProviderName:
_, err := dbHandle.ExecContext(ctx, `SELECT RELEASE_LOCK('sftpgo.migration')`)
if err != nil {
providerLog(logger.LevelWarn, "unable to release lock: %v", err)
} else {
providerLog(logger.LevelInfo, "released database lock")
}
}
}
func sqlCommonExecuteTx(ctx context.Context, dbHandle *sql.DB, txFn func(*sql.Tx) error) error {
if config.Driver == CockroachDataProviderName {
return crdb.ExecuteTx(ctx, dbHandle, nil, txFn)
}
tx, err := dbHandle.BeginTx(ctx, nil)
if err != nil {
return err
}
err = txFn(tx)
if err != nil {
// we don't change the returned error
tx.Rollback() //nolint:errcheck
return err
}
return tx.Commit()
}