// RemoteAddrUsername returns some parts of CallInfo if set
func (stats *LogStats) RemoteAddrUsername() (string, string) {
	ci, ok := callinfo.FromContext(stats.ctx)
	if !ok {
		return "", ""
	}
	return ci.RemoteAddr(), ci.Username()
}

func (qre *QueryExecutor) checkPermissions() {
	// Skip permissions check if we have a background context.
	if qre.ctx == context.Background() {
		return
	}

	// Blacklist
	remoteAddr := ""
	username := ""
	ci, ok := callinfo.FromContext(qre.ctx)
	if ok {
		remoteAddr = ci.RemoteAddr()
		username = ci.Username()
	}
	action, desc := qre.plan.Rules.getAction(remoteAddr, username, qre.bindVars)
	switch action {
	case QR_FAIL:
		panic(NewTabletError(ErrFail, "Query disallowed due to rule: %s", desc))
	case QR_FAIL_RETRY:
		panic(NewTabletError(ErrRetry, "Query disallowed due to rule: %s", desc))
	}

	// Perform table ACL check if it is enabled
	if qre.plan.Authorized != nil && !qre.plan.Authorized.IsMember(username) {
		errStr := fmt.Sprintf("table acl error: %q cannot run %v on table %q", username, qre.plan.PlanId, qre.plan.TableName)
		// Raise error if in strictTableAcl mode, else just log an error
		if qre.qe.strictTableAcl {
			panic(NewTabletError(ErrFail, "%s", errStr))
		}
		qre.qe.accessCheckerLogger.Errorf("%s", errStr)
	}
}

// rpcWrapper handles all the logic for rpc calls.
func (agent *ActionAgent) rpcWrapper(ctx context.Context, name string, args, reply interface{}, verbose bool, f func() error, lock, runAfterAction bool) (err error) {
	defer func() {
		if x := recover(); x != nil {
			log.Errorf("TabletManager.%v(%v) on %v panic: %v\n%s", name, args, topo.TabletAliasString(agent.TabletAlias), x, tb.Stack(4))
			err = fmt.Errorf("caught panic during %v: %v", name, x)
		}
	}()

	from := ""
	ci, ok := callinfo.FromContext(ctx)
	if ok {
		from = ci.Text()
	}

	if lock {
		beforeLock := time.Now()
		agent.actionMutex.Lock()
		defer agent.actionMutex.Unlock()
		if time.Now().Sub(beforeLock) > rpcTimeout {
			return fmt.Errorf("server timeout for " + name)
		}
	}

	if err = f(); err != nil {
		log.Warningf("TabletManager.%v(%v)(on %v from %v) error: %v", name, args, topo.TabletAliasString(agent.TabletAlias), from, err.Error())
		return fmt.Errorf("TabletManager.%v on %v error: %v", name, topo.TabletAliasString(agent.TabletAlias), err)
	}
	if verbose {
		log.Infof("TabletManager.%v(%v)(on %v from %v): %#v", name, args, topo.TabletAliasString(agent.TabletAlias), from, reply)
	}
	if runAfterAction {
		err = agent.refreshTablet(ctx, "RPC("+name+")")
	}
	return
}

// checkPermissions
func (qre *QueryExecutor) checkPermissions() error {
	// Skip permissions check if we have a background context.
	if qre.ctx == context.Background() {
		return nil
	}

	// Blacklist
	remoteAddr := ""
	username := ""
	ci, ok := callinfo.FromContext(qre.ctx)
	if ok {
		remoteAddr = ci.RemoteAddr()
		username = ci.Username()
	}
	action, desc := qre.plan.Rules.getAction(remoteAddr, username, qre.bindVars)
	switch action {
	case QR_FAIL:
		return NewTabletError(ErrFail, "Query disallowed due to rule: %s", desc)
	case QR_FAIL_RETRY:
		return NewTabletError(ErrRetry, "Query disallowed due to rule: %s", desc)
	}

	// a superuser that exempts from table ACL checking.
	if qre.qe.exemptACL == username {
		qre.qe.tableaclExemptCount.Add(1)
		return nil
	}
	tableACLStatsKey := []string{
		qre.plan.TableName,
		// TODO(shengzhe): use table group instead of username.
		username,
		qre.plan.PlanId.String(),
		username,
	}
	if qre.plan.Authorized == nil {
		return NewTabletError(ErrFail, "table acl error: nil acl")
	}
	// perform table ACL check if it is enabled.
	if !qre.plan.Authorized.IsMember(username) {
		if qre.qe.enableTableAclDryRun {
			qre.qe.tableaclPseudoDenied.Add(tableACLStatsKey, 1)
			return nil
		}
		// raise error if in strictTableAcl mode, else just log an error.
		if qre.qe.strictTableAcl {
			errStr := fmt.Sprintf("table acl error: %q cannot run %v on table %q", username, qre.plan.PlanId, qre.plan.TableName)
			qre.qe.tableaclDenied.Add(tableACLStatsKey, 1)
			qre.qe.accessCheckerLogger.Errorf("%s", errStr)
			return NewTabletError(ErrFail, "%s", errStr)
		}
		return nil
	}
	qre.qe.tableaclAllowed.Add(tableACLStatsKey, 1)
	return nil
}

// GetQueryzRows returns a list of QueryDetailzRow sorted by start time
func (ql *QueryList) GetQueryzRows() []QueryDetailzRow {
	ql.mu.Lock()
	rows := []QueryDetailzRow{}
	for _, qd := range ql.queryDetails {
		row := QueryDetailzRow{
			Query:       qd.query,
			ContextHTML: callinfo.FromContext(qd.context).HTML(),
			Start:       qd.start,
			Duration:    time.Now().Sub(qd.start),
			ConnID:      qd.connID,
		}
		rows = append(rows, row)
	}
	ql.mu.Unlock()
	sort.Sort(byStartTime(rows))
	return rows
}

func (qre *QueryExecutor) checkPermissions() {
	// Blacklist
	ci := callinfo.FromContext(qre.ctx)
	action, desc := qre.plan.Rules.getAction(ci.RemoteAddr(), ci.Username(), qre.bindVars)
	switch action {
	case QR_FAIL:
		panic(NewTabletError(FAIL, "Query disallowed due to rule: %s", desc))
	case QR_FAIL_RETRY:
		panic(NewTabletError(RETRY, "Query disallowed due to rule: %s", desc))
	}

	// ACLs
	if !qre.plan.Authorized.IsMember(ci.Username()) {
		errStr := fmt.Sprintf("table acl error: %v cannot run %v on table %v", ci.Username(), qre.plan.PlanId, qre.plan.TableName)
		if qre.qe.strictTableAcl {
			panic(NewTabletError(FAIL, "%s", errStr))
		}
		qre.qe.accessCheckerLogger.Errorf("%s", errStr)
	}
}

// checkPermissions
func (qre *QueryExecutor) checkPermissions() error {
	// Skip permissions check if we have a background context.
	if qre.ctx == context.Background() {
		return nil
	}

	// Blacklist
	remoteAddr := ""
	username := ""
	ci, ok := callinfo.FromContext(qre.ctx)
	if ok {
		remoteAddr = ci.RemoteAddr()
		username = ci.Username()
	}
	action, desc := qre.plan.Rules.getAction(remoteAddr, username, qre.bindVars)
	switch action {
	case QR_FAIL:
		return NewTabletError(ErrFail, vtrpc.ErrorCode_BAD_INPUT, "Query disallowed due to rule: %s", desc)
	case QR_FAIL_RETRY:
		return NewTabletError(ErrRetry, vtrpc.ErrorCode_QUERY_NOT_SERVED, "Query disallowed due to rule: %s", desc)
	}

	callerID := callerid.ImmediateCallerIDFromContext(qre.ctx)
	if callerID == nil {
		if qre.qe.strictTableAcl {
			return NewTabletError(ErrFail, vtrpc.ErrorCode_UNAUTHENTICATED, "missing caller id")
		}
		return nil
	}

	// a superuser that exempts from table ACL checking.
	if qre.qe.exemptACL == callerID.Username {
		qre.qe.tableaclExemptCount.Add(1)
		return nil
	}

	// empty table name, do not need a table ACL check.
	if qre.plan.TableName == "" {
		return nil
	}

	if qre.plan.Authorized == nil {
		return NewTabletError(ErrFail, vtrpc.ErrorCode_PERMISSION_DENIED, "table acl error: nil acl")
	}
	tableACLStatsKey := []string{
		qre.plan.TableName,
		qre.plan.Authorized.GroupName,
		qre.plan.PlanId.String(),
		callerID.Username,
	}
	// perform table ACL check if it is enabled.
	if !qre.plan.Authorized.IsMember(callerID.Username) {
		if qre.qe.enableTableAclDryRun {
			qre.qe.tableaclPseudoDenied.Add(tableACLStatsKey, 1)
			return nil
		}
		// raise error if in strictTableAcl mode, else just log an error.
		if qre.qe.strictTableAcl {
			errStr := fmt.Sprintf("table acl error: %q cannot run %v on table %q", callerID.Username, qre.plan.PlanId, qre.plan.TableName)
			qre.qe.tableaclDenied.Add(tableACLStatsKey, 1)
			qre.qe.accessCheckerLogger.Errorf("%s", errStr)
			return NewTabletError(ErrFail, vtrpc.ErrorCode_PERMISSION_DENIED, "%s", errStr)
		}
		return nil
	}
	qre.qe.tableaclAllowed.Add(tableACLStatsKey, 1)
	return nil
}

func (log *SQLQueryStats) ContextHTML() template.HTML {
	return callinfo.FromContext(log.context).HTML()
}

func (log *SQLQueryStats) Username() string {
	return callinfo.FromContext(log.context).Username()
}

func (log *SQLQueryStats) RemoteAddr() string {
	return callinfo.FromContext(log.context).RemoteAddr()
}